FastAPI中SQLAlchemy动态切换多Schema获取数据的问题咨询

直接修改models.DBStatus.__table__.schema属于全局修改表元数据，而SQLAlchemy的Session会缓存表结构信息，第一次查询后后续查询不会重新读取修改后的schema，导致始终使用第一个设置的schema。另外这种全局修改在并发请求场景下还会引发数据混乱的问题。

不需要重新创建Engine，Engine是连接池，复用更高效。可以通过以下方式动态指定schema查询：

方法1：动态生成带指定schema的表对象 #

每次查询时基于原表元数据生成带目标schema的临时表对象，不修改全局模型：

def fetch_all_db_status(session: Session, schema_names):
    db_status = []
    try:
        for schema_name in schema_names:
            # 基于原表生成带指定schema的临时表对象
            target_table = models.DBStatus.__table__.tometadata(
                session.get_bind().metadata, schema=schema_name
            )
            # 执行查询，返回Row对象
            rows = session.query(target_table).all()
            # 可选：将Row对象转换为DBStatus模型实例
            status_instances = [models.DBStatus(**row._asdict()) for row in rows]
            db_status.append(status_instances)
    except Exception as ee:
        logger.error(f"Error while fetching data from DB_STATUS for schema: {schema_name}")
        logger.error(str(ee))
    return db_status

方法2：使用SQLAlchemy的select语句直接指定schema（SQLAlchemy 2.0+） #

如果使用SQLAlchemy 2.0及以上版本，可以用更简洁的select语法：

from sqlalchemy import select

def fetch_all_db_status(session: Session, schema_names):
    db_status = []
    try:
        for schema_name in schema_names:
            # 直接在select中指定表的schema
            stmt = select(models.DBStatus).select_from(
                models.DBStatus.__table__.schema(schema_name)
            )
            result = session.execute(stmt).scalars().all()
            db_status.append(result)
    except Exception as ee:
        logger.error(f"Error while fetching data from DB_STATUS for schema: {schema_name}")
        logger.error(str(ee))
    return db_status

• 绝对不要修改全局模型的__table__.schema，尤其是在FastAPI这种并发场景下，会导致不同请求之间的schema相互干扰。
• 如果你的数据库支持schema切换（比如PostgreSQL的SET search_path），也可以在每次查询前执行session.execute(text(f"SET search_path TO {schema_name}"))，但这种方法需要注意Session的隔离性，避免影响后续查询。

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