我有许多SQL表需要用新信息更新。我正在使用Python和ODBC制作一个脚本,自动更新所有的表。我最初的想法是截断每个表,然后使用更新的源文件重新加载它们,但由于它们的外键限制,我无法截断它们。我试着打开外键约束,但它仍然不允许我截断。
我正在寻找更新这些表格的最佳方式。我有SQL文件中的源数据,我正在加载到pandas数据帧中,我不知道在源文件的每次更新中哪些数据会发生变化(如果有任何变化的话(,它可以添加行、删除行,或者只更改一些数量。有人对这种情况有什么建议吗?
如本答案所述,我们可以使用关闭FK检查
ALTER TABLE tableName NOCHECK CONSTRAINT ALL
并使用重新启用
ALTER TABLE tableName WITH CHECK CHECK CONSTRAINT ALL
测试表明,这不允许我们执行TRUNCATE tableName
或DROP TABLE tableName
,可能是因为这些操作的代码只是检查FK是否存在,而不是当前是否正在强制执行。然而,它确实允许我们进行DELETE FROM tableName
# fmt:off
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey
from sqlalchemy.exc import IntegrityError
# fmt:on
engine = create_engine("mssql+pyodbc://scott:tiger^5HHH@mssql_199")
meta = MetaData()
# example environment
parent = Table(
"parent",
meta,
Column("id", Integer, primary_key=True, autoincrement=False),
Column("name", String(50)),
)
child = Table(
"child",
meta,
Column("id", Integer, primary_key=True, autoincrement=False),
Column("name", String(50)),
Column("parent_id", Integer, ForeignKey("parent.id")),
)
meta.drop_all(engine, checkfirst=True)
meta.create_all(engine)
# example data
with engine.begin() as conn:
conn.execute(parent.insert(), dict(id=1, name="parent_1"))
conn.execute(child.insert(), dict(id=11, name="child_1", parent_id=1))
# try delete without disabling FK check
with engine.begin() as conn:
try:
conn.exec_driver_sql("DELETE FROM parent")
except IntegrityError as e:
print(e)
# (pyodbc.IntegrityError) ('23000', '[23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]The DELETE statement conflicted with the REFERENCE constraint "FK__child__parent_id__7B5B524B". The conflict occurred in database "test", table "dbo.child", column 'parent_id'. (547) (SQLExecDirectW); [23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]The statement has been terminated. (3621)')
# try disabling FK check
with engine.begin() as conn:
conn.exec_driver_sql("ALTER TABLE child NOCHECK CONSTRAINT ALL")
conn.exec_driver_sql("DELETE FROM parent")
conn.exec_driver_sql(
"INSERT INTO parent (id, name) VALUES (1, 'new_parent_1')"
)
conn.exec_driver_sql("ALTER TABLE child WITH CHECK CHECK CONSTRAINT ALL")
# (no errors)
# verify results
with engine.begin() as conn:
print(conn.exec_driver_sql("SELECT * FROM parent").all())
# [(1, 'new_parent_1')]