使用Python导出PostgreSQL数据库架构

Question

我正在寻找一种使用Python导出PostgreSQL数据库模式的方法。理想情况下，结果应该是一个.sql导出文件或任何其他格式，稍后可以由SQLAlchemy使用以从该模式创建新数据库。

我用SQLAlchemy的MetaData类进行了这个实验：

from sqlalchemy import MetaData, create_engine

engine = create_engine(source_database_url)
test_engine = create_engine(test_database_url)
metadata = MetaData()
metadata.reflect(engine)
metadata.create_all(test_engine)

这段文字的大意是：除了导出模式以供后续使用外，它完全符合我的要求。因此，考虑到SQLAlchemy可以成功地反映模式并创建基于其的另一个表，我希望有一种方法在该过程中以编程方式导出它。

我知道我可以从代码中调用pg_dump和pg_restore，但我想避免外部依赖和随之而来的麻烦。

是否有一种使用SQLAlchemy或其他Python库实现此目的的方法？

- Georgiana.b

Answer 1

进一步调研后，我发现 SQLAlchemy 文档中这个 有用的部分。

所以，从那段代码片段开始，我创建了以下类：

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

class SchemaDump(object):
    def __init__(self, db_url, schema_file_path):
        self.db_url = db_url
        self.schema_file_path = schema_file_path
        self.buf = io.BytesIO()

    def dump_shema(self):
        engine = create_engine(self.db_url)
        metadata = MetaData()
        metadata.reflect(engine)

        def dump(sql, *multiparams, **params):
            f = sql.compile(dialect=engine.dialect)
            self.buf.write(str(f).encode('utf-8'))
            self.buf.write(b';\n')

        new_engine = create_engine(self.db_url, strategy='mock', executor=dump)
        metadata.create_all(new_engine, checkfirst=True)

        with io.open(self.schema_file_path, 'wb+') as schema:
            schema.write(self.buf.getvalue())

这还很初步，但主要思路是在buf中捕获由sql.compile(dialect = engine.dialect)返回的原始SQL语句，并将它们写入文件。 我编写了一个同样简陋的类来从上述类创建的.sql转储中还原数据库。

import io
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class RestoreSchema(object):
    def __init__(self, db_url, schema_file_path):
        self.db_url = db_url
        self.schema_file_path = schema_file_path

    def restore_schema(self):
        raw_schema = ''
        with io.open(self.schema_file_path) as sql_schema:
            raw_schema = sql_schema.read()
        engine = create_engine(self.db_url)
        Session = sessionmaker(bind=engine)
        session = Session()
        conn = engine.connect()
        transaction = conn.begin()
        try:
            conn.execute(raw_schema)
            transaction.commit()
        except Exception as e:
            transaction.rollback()
            raise e
        finally:
            session.close()

您仍然需要担心表是否已经存在等问题，但效果与我提问中的代码片段完全相同。

- Georgiana.b