如何在FastAPI中通过URL前缀切换数据库实例？

1. 抽离数据库引擎创建逻辑 #

先把固定的引擎初始化改成可接收master_i参数的动态逻辑，通过配置映射匹配不同实例的连接信息：

# 维护各主实例的数据库连接配置（可从.env或配置文件加载）
MASTER_CONFIG_MAP = {
    "master1": {"user": "dev_user", "password": "dev_pass", "host": "dev-host", "port": 5432},
    "master2": {"user": "test_user", "password": "test_pass", "host": "test-host", "port": 5432},
    "master3": {"user": "prod_user", "password": "prod_pass", "host": "prod-host", "port": 5432},
}

def get_sqlalchemy_engine(master_name: str, db_name: str):
    # 校验master实例合法性
    if master_name not in MASTER_CONFIG_MAP:
        raise ValueError(f"无效的主数据库实例: {master_name}")
    config = MASTER_CONFIG_MAP[master_name]
    # 拼接PostgreSQL连接字符串
    conn_str = f"postgresql://{config['user']}:{config['password']}@{config['host']}:{config['port']}/{db_name}"
    return create_engine(conn_str)

2. 复用路由逻辑（以FastAPI为例） #

将原有路由封装为独立的子路由，通过路由前缀挂载和依赖注入实现多路径兼容，无需重复编写接口代码：

步骤1：封装原有业务路由 #

把/db_j/{tables_xyz}的逻辑抽离到子路由中，通过依赖获取当前使用的master实例：

from fastapi import APIRouter, Depends, Path

# 子路由：封装所有数据库表级接口逻辑
db_table_router = APIRouter()

# 定义依赖：获取master实例名，兼容原无前缀接口（默认用master1）
def get_current_master(master_name: str = Path(None, alias="master_i")):
    return master_name or "master1"

# 原有接口逻辑直接复用，仅新增master参数
@db_table_router.get("/{db_name}/table1/")
def fetch_table1_data(
    db_name: str,
    master_name: str = Depends(get_current_master)
):
    engine = get_sqlalchemy_engine(master_name, db_name)
    # 原接口业务逻辑（示例：查询table1数据）
    with engine.connect() as conn:
        result = conn.execute("SELECT * FROM table1 LIMIT 10")
        return {
            "data": [dict(row) for row in result],
            "used_master": master_name,
            "used_db": db_name
        }

步骤2：挂载多版本路由 #

在主应用中同时挂载带master前缀和不带前缀的路由，实现两种路径兼容：

from fastapi import FastAPI

app = FastAPI()

# 挂载带master前缀的路由：支持/{master_i}/db_j/table1/
app.include_router(db_table_router, prefix="/{master_i}", tags=["多实例接口"])
# 挂载原无前缀路由：保持/db_j/table1/路径可用
app.include_router(db_table_router, tags=["原兼容接口"])

3. Flask框架兼容方案（可选） #

如果使用Flask，可通过前置钩子解析路径并传递master实例，无需修改原有路由：

from flask import Flask, g, request

app = Flask(__name__)

@app.before_request
def resolve_master_instance():
    path_segments = request.path.lstrip("/").split("/")
    # 判断路径是否包含合法的master前缀
    if path_segments and path_segments[0] in MASTER_CONFIG_MAP:
        g.master_name = path_segments[0]
        # 重写请求路径，让原有路由匹配/db_j/table1/
        request.path = "/" + "/".join(path_segments[1:])
    else:
        g.master_name = "master1"  # 默认实例

# 原有路由逻辑完全复用
@app.route("/<db_name>/table1/")
def get_table1_data(db_name):
    engine = get_sqlalchemy_engine(g.master_name, db_name)
    # 原业务逻辑...
    with engine.connect() as conn:
        result = conn.execute("SELECT * FROM table1 LIMIT 10")
        return {"data": [dict(row) for row in result], "master": g.master_name}

关键注意事项 #

• 必须对master_i参数做合法性校验，避免无效实例导致连接失败
• 异步场景需使用SQLAlchemy异步引擎（AsyncEngine）
• 默认实例可通过配置文件设置，避免硬编码
• 接口文档（如FastAPI Swagger）会自动生成两种路径的接口，便于测试

内容的提问来源于stack exchange，提问作者d.eng 489