让 SQLAlchemy 在 create_all 上发布 CREATE SCHEMA

Posted 2023-03-22

【中文标题】让 SQLAlchemy 在 create_all 上发布 CREATE SCHEMA

【英文标题】：Getting SQLAlchemy to issue CREATE SCHEMA on create_all

【发布时间】：2012-11-20 14:18:11

【问题描述】：

我有一个带有如下架构参数的 SqlAlchemy 模型：

Base = declarative_base()

class Road(Base):
  __tablename__ = "roads"
  __table_args__ = 'schema': 'my_schema'
  id = Column(Integer, primary_key=True)

当我使用 Base.metadata.create_all(engine) 时，它会正确地发出 CREATE TABLE 前面的架构名称，就像 CREATE TABLE my_schema.roads ( 一样，但 Postgresql 正确地抱怨架构不存在。

我是否错过了让 SqlAlchemy 发出 CREATE SCHEMA my_schema 的步骤，还是我必须手动调用它？

【参考方案1】：

我已经在我的数据库初始化脚本上手动完成了，如下所示：

from sqlalchemy.schema import CreateSchema
engine.execute(CreateSchema('my_schema'))

但这似乎没有我预期的那么神奇。

【参考方案2】：

我遇到了同样的问题，并认为发布 DDL 的“最干净”的方式是这样的：

from sqlalchemy import event
from sqlalchemy.schema import CreateSchema

event.listen(Base.metadata, 'before_create', CreateSchema('my_schema'))

这将确保在创建基础元数据中包含的任何内容之前，您拥有它的架构。但是，这不会检查架构是否已经存在。

如果您懒得写check_schema 回调（文档中should_create 上的“Controlling DDL Sequences”），您可以使用CreateSchema('my_schema').execute_if(callback_=check_schema)。或者，作为一种简单的方法，只需使用 DDL("CREATE SCHEMA IF NOT EXISTS my_schema") 代替（对于 Postgres）：

from sqlalchemy import DDL

event.listen(Base.metadata, 'before_create', DDL("CREATE SCHEMA IF NOT EXISTS my_schema"))

【讨论】：

我可以只为 Postgres 中新创建的架构运行迁移吗？

@Emu 您可能想将其作为一个单独的问题提出。但是您可能必须检查架构是否不存在 - 如果不存在，请创建它并运行您想要的任何迁移。我的回答并未严格涵盖这一点。您可能想查看完整的迁移库，例如SQLAlchemy 迁移或 alembic。

【参考方案3】：

我编写了一个函数，它根据接受的答案创建声明的模式。它使用每个映射类的 __table_args__ dict 中的 schema 值。

from sqlalchemy import event, DDL


# Import or write your mapped classes and configuration here

def init_db():
    for mapper in Base.registry.mappers:
        cls = mapper.class_
        if issubclass(cls, Base):
            table_args = getattr(cls, '__table_args__', None)
            if table_args:
                schema = table_args.get('schema')
                if schema:
                    stmt = f"CREATE SCHEMA IF NOT EXISTS schema"
                    event.listen(Base.metadata, 'before_create', DDL(stmt))
    Base.metadata.create_all(bind=engine)