如何基于FastAPI端点参数动态修改SQLAlchemy数据库连接字符串？

当前代码中SQLAlchemy固定连接patients库，通过get_db依赖提供会话，希望在FastAPI端点中通过{x}参数动态切换目标库（比如x=05时连接patients_05），但端点参数无法直接传递给get_db，以下是可行解决方案：

原代码参考 #

import urllib.parse
from sqlalchemy import create_engine, declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends, FastAPI, HTMLResponse

params = urllib.parse.quote_plus("DRIVER={ODBC Driver 11 for SQL Server};"
                                 "SERVER=DESKTOP-90423MF;"
                                 "DATABASE=patients;" 
                                 "UID=sa;"
                                 "PWD=randompass")

engine = create_engine("mssql+pyodbc:///?odbc_connect={}".format(params))
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 端点代码
app = FastAPI()
@app.get("/{x}/{patient_id}/{file_no}/", response_class=HTMLResponse)
async def get_patient_report(x: str, patient_id: str, file_no: str, db: Session = Depends(get_db)):
    # 业务逻辑
    pass

解决方案 #

1. 重构连接配置，添加引擎缓存 #

抽离固定连接参数，动态生成数据库名，并缓存已创建的引擎，避免重复初始化：

# 基础连接模板（移除固定的DATABASE字段）
BASE_CONN_TEMPLATE = (
    "DRIVER={ODBC Driver 11 for SQL Server};"
    "SERVER=DESKTOP-90423MF;"
    "UID=sa;"
    "PWD=randompass"
)

# 缓存已创建的引擎，减少重复开销
engine_cache = {}

def get_engine(db_suffix: str):
    # 生成目标数据库名
    db_name = f"patients_{db_suffix}" if db_suffix else "patients"
    # 缓存命中直接返回
    if db_name in engine_cache:
        return engine_cache[db_name]
    # 生成新连接并缓存
    conn_str = f"{BASE_CONN_TEMPLATE};DATABASE={db_name}"
    quoted_params = urllib.parse.quote_plus(conn_str)
    engine = create_engine(f"mssql+pyodbc:///?odbc_connect={quoted_params}")
    engine_cache[db_name] = engine
    return engine

2. 修改get_db为带参数的依赖 #

让get_db接收db_suffix参数，动态生成对应数据库的会话：

def get_db(db_suffix: str):
    engine = get_engine(db_suffix)
    # 针对当前引擎创建会话工厂
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. 端点中传递参数给依赖 #

通过lambda将端点的x参数传入get_db依赖：

@app.get("/{x}/{patient_id}/{file_no}/", response_class=HTMLResponse)
async def get_patient_report(x: str, patient_id: str, file_no: str, db: Session = Depends(lambda: get_db(x))):
    # 此时db已指向patients_x库的会话，可直接执行业务逻辑
    # 示例：patient = db.query(PatientModel).filter(PatientModel.id == patient_id).first()
    return HTMLResponse(content=f"已连接至patients_{x}数据库")

可选：参数合法性校验 #

如果需要限制x的格式（比如必须是两位数字），可以用Pydantic的校验规则：

from pydantic import constr

@app.get("/{x}/{patient_id}/{file_no}/", response_class=HTMLResponse)
async def get_patient_report(
    x: constr(regex=r"^\d{2}$"),  # 限制x为两位数字
    patient_id: str, 
    file_no: str, 
    db: Session = Depends(lambda: get_db(x))
):
    # 业务逻辑
    pass

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