如何通过SQLAlchemy为PostgreSQL表动态指定Schema？

我尝试通过以下代码创建名为"networks"的表：

import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    __table_args__ = {'schema': ''}

    def __init__(self, schema, **kw):
        super().__init__(**kw)
        Base.__table_args__['schema'] = schema

class DBNetwork(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))

if __name__ == '__main__':
    engine = sa.create_engine('postgresql+psycopg2://postgres:pass@localhost/test', 
                              pool_size=20, max_overflow=0)
    Base('test')

    with engine.connect() as connection:
        connection.execute(sa.schema.CreateSchema(
            Base.__table_args__['schema'], if_not_exists=True))
        connection.commit()
    Base.metadata.create_all(self.engine)

我原本期望：定义Base类时设置__table_args__ = {'schema': ''}，未指定schema时在public schema创建表，调用时可指定schema创建表。但实际发现Base.metadata已基于DBNetwork初始继承的__table_args__ = {'schema': ''}生成，无法识别通过Base('test')进行的Schema选择。

此外，调用Base('test')后打印DBNetwork.__table_args__显示为：

print(DBNetwork.__table_args__)
{'schema': 'test'}

请问如何修改代码实现Schema选择并在该Schema创建表？我不想为每个Base子类硬编码__table_args__ = {'schema': 'test'}。

问题根源在于：DeclarativeBase的子类（比如DBNetwork）在定义时，就已经将自身的表结构注册到Base.metadata中，此时会读取当时Base.__table_args__的schema值。后续修改Base.__table_args__只会改变类属性，但不会更新已经注册到metadata里的表的schema配置。

以下是两种可行的修改方案：

方案一：先指定schema再定义子类 #

调整代码顺序，先根据指定的schema创建自定义Base类，再定义表模型，从根源上确保模型绑定正确的schema：

import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

def create_base_with_schema(schema=None):
    # 创建带指定schema的Base类，未指定则默认使用public schema
    class Base(DeclarativeBase):
        __table_args__ = {'schema': schema} if schema else {}
    return Base

if __name__ == '__main__':
    engine = sa.create_engine('postgresql+psycopg2://postgres:pass@localhost/test', 
                              pool_size=20, max_overflow=0)
    target_schema = 'test'
    # 先创建带目标schema的Base类
    Base = create_base_with_schema(target_schema)

    # 再定义表模型，继承这个配置好的Base
    class DBNetwork(Base):
        __tablename__ = 'networks'

        id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
        name: Mapped[str] = mapped_column(sa.String(60))

    # 创建目标schema（如果不存在）
    with engine.connect() as connection:
        connection.execute(sa.schema.CreateSchema(target_schema, if_not_exists=True))
        connection.commit()
    
    # 创建表
    Base.metadata.create_all(engine)

方案二：动态更新已注册表的schema #

如果无法调整模型定义顺序，可以在创建表前，遍历Base.metadata中的所有表，手动更新它们的schema属性：

import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    __table_args__ = {'schema': ''}

class DBNetwork(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))

if __name__ == '__main__':
    engine = sa.create_engine('postgresql+psycopg2://postgres:pass@localhost/test', 
                              pool_size=20, max_overflow=0)
    target_schema = 'test'
    # 设置Base类的schema属性
    Base.__table_args__['schema'] = target_schema

    # 创建目标schema（如果不存在）
    with engine.connect() as connection:
        connection.execute(sa.schema.CreateSchema(target_schema, if_not_exists=True))
        connection.commit()
    
    # 遍历所有已注册的表，手动更新它们的schema
    for table in Base.metadata.tables.values():
        table.schema = target_schema
    
    # 创建表
    Base.metadata.create_all(engine)

方案说明 #

• 方案一更符合SQLAlchemy的设计逻辑，模型定义时就明确了所属schema，避免元数据不一致的问题。
• 方案二适合无法调整模型定义顺序的场景，通过手动修正元数据实现schema指定。

内容的提问来源于stack exchange，提问作者oakca