これは、トランザクション内のすべての制約を削除し、いくつかのクエリを実行してから、それらすべての制約を再作成する Python スクリプトです。 pg_get_constraintdef
これは非常に簡単です:
class no_constraints(object):
def __init__(self, connection):
self.connection = connection
def __enter__(self):
self.transaction = self.connection.begin()
try:
self._drop_constraints()
except:
self.transaction.rollback()
raise
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None:
self.transaction.rollback()
else:
try:
self._create_constraints()
self.transaction.commit()
except:
self.transaction.rollback()
raise
def _drop_constraints(self):
self._constraints = self._all_constraints()
for schemaname, tablename, name, def_ in self._constraints:
self.connection.execute('ALTER TABLE "%s.%s" DROP CONSTRAINT %s' % (schemaname, tablename, name))
def _create_constraints(self):
for schemaname, tablename, name, def_ in self._constraints:
self.connection.execute('ALTER TABLE "%s.%s" ADD CONSTRAINT %s %s' % (schamename, tablename, name, def_))
def _all_constraints(self):
return self.connection.execute("""
SELECT n.nspname AS schemaname, c.relname, conname, pg_get_constraintdef(r.oid, false) as condef
FROM pg_constraint r, pg_class c
LEFT JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE r.contype = 'f'
and r.conrelid=c.oid
""").fetchall()
if __name__ == '__main__':
# example usage
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@host/dbname', echo=True)
conn = engine.connect()
with no_contraints(conn):
r = conn.execute("delete from table1")
print "%d rows affected" % r.rowcount
r = conn.execute("delete from table2")
print "%d rows affected" % r.rowcount