FastAPI+SQLAlchemy启动时创建PostgreSQL Schema与表的问题

一、正确创建Schema的方法 #

针对你遇到的Schema创建失败问题，以下两种可行方案：

方案1：通过run_sync执行Schema创建逻辑 #

利用run_sync传入一个接收同步连接的函数，在函数内完成Schema创建和表初始化：

from sqlalchemy import CreateSchema
from sqlalchemy.ext.asyncio import create_async_engine

async def migrate_tables() -> None:
    engine = create_async_engine(api_config.DB_URI)
    async with engine.begin() as connection:
        logger.info("CREATING SCHEMA")
        # 传入处理Schema和表创建的同步函数
        await connection.run_sync(create_schema_and_tables)
    logger.info("FINISHED CREATING SCHEMA AND TABLES")

def create_schema_and_tables(sync_conn):
    schema_name = 'my_schema'
    # 创建Schema，带if_not_exists避免重复创建报错
    sync_conn.execute(CreateSchema(schema_name, if_not_exists=True))
    sync_conn.commit()
    # 在指定Schema下创建所有表
    Base.metadata.create_all(sync_conn)

方案2：执行原生SQL语句创建Schema #

直接通过异步连接执行原生PostgreSQL语句，更直观：

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

async def migrate_tables() -> None:
    engine = create_async_engine(api_config.DB_URI)
    async with engine.begin() as connection:
        logger.info("CREATING SCHEMA")
        schema_name = 'my_schema'
        # 执行原生SQL创建Schema
        await connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_name}"))
        logger.info("CREATING TABLES")
        await connection.run_sync(Base.metadata.create_all)
    logger.info("FINISHED CREATING SCHEMA AND TABLES")

你之前的错误原因分析 #

1. 同步方法未生效：你的create_schema函数中错误使用了yield，将普通函数变成了生成器函数，调用时不会自动执行内部逻辑，导致Schema未创建。
2. 直接传CreateSchema给run_sync报错：run_sync需要接收一个函数（参数为同步连接），而你直接传入了CreateSchema实例，它需要绑定连接才能执行，因此抛出missing required positional argument: 'bind'错误。

二、表初始化方式的优化建议 #

1. 复用数据库引擎 #

避免每次执行migrate_tables时都创建新引擎，全局初始化引擎可减少连接池重复创建的开销：

# 在模块全局初始化引擎
engine = create_async_engine(api_config.DB_URI)

async def migrate_tables() -> None:
    async with engine.begin() as connection:
        await connection.execute(text("CREATE SCHEMA IF NOT EXISTS my_schema"))
        await connection.run_sync(Base.metadata.create_all)

2. 添加异常处理 #

捕获数据库操作异常，记录详细日志，避免应用启动失败时无有效排查信息：

async def migrate_tables() -> None:
    try:
        async with engine.begin() as connection:
            logger.info("STARTING DATABASE INITIALIZATION")
            await connection.execute(text("CREATE SCHEMA IF NOT EXISTS my_schema"))
            await connection.run_sync(Base.metadata.create_all)
        logger.info("DATABASE INITIALIZATION SUCCESSFUL")
    except Exception as e:
        logger.error(f"DATABASE INITIALIZATION FAILED: {str(e)}")
        raise  # 可根据需求选择是否终止应用启动

3. 生产环境建议使用Alembic做迁移 #

create_all仅能处理表的首次创建，无法应对表结构变更场景。生产环境推荐使用SQLAlchemy官方迁移工具Alembic，实现版本化的数据库迁移，启动时可自动执行未应用的迁移脚本，比create_all更可靠。

4. 模型显式指定Schema #

确保你的SQLAlchemy模型中明确指定目标Schema，避免表创建到默认publicSchema：

class MyModel(Base):
    __tablename__ = 'my_table'
    __table_args__ = {'schema': 'my_schema'}
    id = Column(Integer, primary_key=True)
    # 其他字段...

5. 控制初始化执行时机 #

在生产环境，可通过配置开关控制是否在启动时执行初始化，避免每次重启都重复执行无意义的Schema/表检查：

async def lifespan(api: FastAPI):
    if api_config.RUN_MIGRATIONS_ON_STARTUP:
        await migrate_tables()
    yield

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