SQLAlchemy动态Schema名称在try/except或with块中失效问题

在SQLAlchemy中使用多Schema时，发现通过上下文管理器（with-block）或try/except-block返回会话时，动态指定的Schema名称无法正常生效，执行数据库操作时仍使用默认Schema名称，导致报错。

执行示例代码 #

def import_units(schema_name: str):
    from taxes.models import Tax
    
    dbs = get_db_session(schema_name)
    dbs.add(Tax(percentage=19.00, abbreviation="abc"))
    dbs.commit()  # 提交时出现错误：
       # sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedTable) relation "per_user.taxes" does not exist
       # LINE 1: INSERT INTO per_user.taxes (percentage, abbreviation, de...

会话生成方式及对应效果 #

def get_db_session(schema_name: str):    
    dbs = SessionLocal()
    dbs.connection(execution_options={"schema_translate_map": {'per_user': schema_name}})
    
    return dbs # 此返回方式生效（可正确转换Schema名称）

    try:
        return dbs # 此方式不生效（未转换，仍使用'per_user'作为Schema名称）
    finally:
        dbs.close()

    with SessionLocal() as s:
        s.connection(execution_options={"schema_translate_map": {'per_user': schema_name}})
        return s # 此方式不生效（未转换，仍使用'per_user'作为Schema名称）

• 直接创建会话并设置连接选项后返回：正常生效，Schema名称被正确替换
• 嵌套在try/finally块中返回：不生效，仍使用原Schema名称per_user
• 通过with上下文管理器创建后返回：不生效，仍使用原Schema名称per_user

基础配置代码 #

from config.settings import SQL_DATABASE_URI
import warnings
from sqlalchemy import create_engine, exc
from sqlalchemy.orm import sessionmaker


engine = create_engine(SQL_DATABASE_URI, pool_pre_ping=True, future=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)
warnings.filterwarnings("always", category=exc.RemovedIn20Warning)

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