更新时间:2023-02-02 22:54:38
我从未在python中找到DROP CASCADE的实现所以我最终写了我自己的。以下是它的外观:
def delete_table(self,table_name,cascade = True):
删除表'table_name'
params =(self.quote_name(table_name),)
如果级联:
conn = self._get_connection()
#获取相关表的列表
sql =SELECT T1.TABLE_NAME \
FROM INFORMATION_SCHEMA.CONSTRAINT_TABLE_USAGE AS T1 \
JOIN SYS.FOREIGN_KEYS AS F \
ON(F.parent_object_id = OBJECT_ID(N'{0}')OR \
F.referenced_object_id = OBJECT_ID(N'{0}'))AND \
T1 .CONSTRAINT_NAME = OBJECT_NAME(F.object_id)
related_tables = self.execute(sql.format(params [0]))
#删除所有约束
constraints = self.execute(SELECT CONSTRAINT_NAME \
FROM INFO RMATION_SCHEMA.TABLE_CONSTRAINTS \
WHERE TABLE_NAME ='{0}'AND \
CONSTRAINT_TYPE ='FOREIGN KEY';。format(table_name))
sql =ALTER TABLE {0} DROP CONSTRAINT {1};
约束约束:
self.execute(sql.format(params [0],constraint [0]))
相关表中的表:
self .delete_table(table [0],cascade)
sql =IF EXISTS(SELECT * \
FROM sys.objects \
WHERE object_id = OBJECT_ID(N' 0}')AND \
type in(N'U'))\
DROP TABLE {0}
self.execute(sql.format(params [ 0]))
else:
self.execute('DROP TABLE%s;'%params)
I'm running a South migration in a Django project that uses Sql Server and pyodbc. This is backwards migration so the South is trying to delete a few of my tables.
The South executes the following method in order to drop the tables:
def delete_table(self, table_name, cascade=True):
"""
Deletes the table 'table_name'.
"""
params = (self.quote_name(table_name), )
if cascade:
self.execute('DROP TABLE %s CASCADE;' % params)
else:
self.execute('DROP TABLE %s;' % params)
drop_table = alias('delete_table')
The problem is that the Sql Server does not support cascade drops, so the migration fails with the following error:
pyodbc.ProgrammingError: ('42000', "[42000] [Microsoft][ODBC SQL Server Driver][SQL Server]Incorrect syntax near the keyword 'CASCADE'. (156) (SQLExecDirectW)")
I want to write to a patch for the South so that it works with the Sql Server. What would be the best way to simulate a DROP CASCADE? I think that both solutions using python or pure SQL would be valid.
I never found an implementation of the DROP CASCADE in python, so I ended up writing my own. Here is how it looks:
def delete_table(self, table_name, cascade=True):
"""
Deletes the table 'table_name'.
"""
params = (self.quote_name(table_name), )
if cascade:
conn = self._get_connection()
# Get a list of related tables
sql = "SELECT T1.TABLE_NAME \
FROM INFORMATION_SCHEMA.CONSTRAINT_TABLE_USAGE AS T1 \
JOIN SYS.FOREIGN_KEYS AS F \
ON (F.parent_object_id = OBJECT_ID(N'{0}') OR \
F.referenced_object_id = OBJECT_ID(N'{0}')) AND \
T1.CONSTRAINT_NAME = OBJECT_NAME(F.object_id)"
related_tables = self.execute(sql.format(params[0]))
# Drop all the constraints
constraints = self.execute("SELECT CONSTRAINT_NAME \
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS \
WHERE TABLE_NAME='{0}' AND \
CONSTRAINT_TYPE='FOREIGN KEY';".format(table_name))
sql = "ALTER TABLE {0} DROP CONSTRAINT {1};"
for constraint in constraints:
self.execute(sql.format(params[0], constraint[0]))
for table in related_tables:
self.delete_table(table[0], cascade)
sql = "IF EXISTS (SELECT * \
FROM sys.objects \
WHERE object_id = OBJECT_ID(N'{0}') AND \
type in (N'U')) \
DROP TABLE {0}"
self.execute(sql.format(params[0]))
else:
self.execute('DROP TABLE %s;' % params)