使用 SQLAlchemy 将一个数据库复制到另一个答案

【问题标题】：Copy one database to another using SQLAlchemy使用 SQLAlchemy 将一个数据库复制到另一个
【发布时间】：2022-01-20 07:37:15
【问题描述】：

我正在尝试使用 SQLAlchemy 复制数据库。第一次尝试是：

from from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker

from urls import engine_urls 

engine1 = create_engine(engine_urls[0])
engine2 = create_engine(engine_urls[1])

metadata = MetaData()
metadata.reflect(engine1)
tables = metadata.tables
metadata.create_all(engine2)


Session1 = sessionmaker(bind=engine1)

from sqlalchemy import insert
 
with Session1.begin() as session:
    for key in tables:
        table_object = tables[key]
        for row in session.query(table_object):
            s = insert(table_object).\
                values(**dict(zip(row.keys(), row))) 
            engine2.execute(s)

但是这段代码不起作用，因为插入的顺序是任意的，这违反了 FK 约束。例如，在父母之前插入一个孩子会导致这种违规行为。我怎样才能完成这项任务？框架中是否有一部分可以轻松做到这一点？没找到。

【问题讨论】：

docs.sqlalchemy.org/en/14/core/…

标签： python sqlalchemy

【解决方案1】：

这是我使用的。效果很好。

from sqlalchemy import create_engine, MetaData, event
from sqlalchemy.sql import sqltypes

#  Requires SQLALCHEMY 1.4+

src_engine = create_engine("sqlite:///mydb.sqlite")
src_metadata = MetaData(bind=src_engine)
exclude_tables = ('sqlite_master', 'sqlite_sequence', 'sqlite_temp_master')

tgt_engine = create_engine("postgresql+psycopg2://@localhost/ngas")
tgt_metadata = MetaData(bind=tgt_engine)

@event.listens_for(src_metadata, "column_reflect")
def genericize_datatypes(inspector, tablename, column_dict):
   column_dict["type"] = column_dict["type"].as_generic(allow_nulltype=True)     

tgt_conn = tgt_engine.connect()
tgt_metadata.reflect()

# drop all tables in target database
for table in reversed(tgt_metadata.sorted_tables):
   if table.name not in exclude_tables:
      print('dropping table =', table.name)
      table.drop()

# # Delete all data in target database
# for table in reversed(tgt_metadata.sorted_tables):
#    table.delete()

tgt_metadata.clear()
tgt_metadata.reflect()
src_metadata.reflect()

# create all tables in target database
for table in src_metadata.sorted_tables:
   if table.name not in exclude_tables:
      table.create(bind=tgt_engine)

# refresh metadata before you can copy data
tgt_metadata.clear()
tgt_metadata.reflect()

# Copy all data from src to target
for table in tgt_metadata.sorted_tables:
   src_table = src_metadata.tables[table.name]
   stmt = table.insert()
   for index, row in enumerate(src_table.select().execute()):
      print("table =", table.name, "Inserting row", index)
      stmt.execute(row._asdict())

【讨论】：