如何在FastAPI中动态切换数据库？（基于Tortoise ORM场景）

完全可以实现类似Django中线程局部存储切换数据库的需求，下面是具体的实现方案：

1. 线程局部存储管理当前数据库 #

先通过threading.local()创建线程局部变量，用来存储当前请求对应的用户专属数据库别名：

import threading

THREAD_LOCAL = threading.local()

def set_current_db(db_name: str):
    THREAD_LOCAL.db_name = db_name

def get_current_db() -> str:
    # 默认使用公共数据库（default）
    return getattr(THREAD_LOCAL, 'db_name', 'default')

2. 请求入口解析Token并设置数据库 #

在API的请求入口（比如FastAPI的依赖、Flask的前置钩子）解析Token，提取用户信息后获取对应的数据库名称，调用set_current_db存入线程局部：

# 以FastAPI为例，自定义依赖
from fastapi import Depends, HTTPException

def parse_token(token: str):
    # 这里实现你的Token解析逻辑，返回用户信息
    ...

def get_user_db_name(user_info):
    # 根据用户信息生成/获取专属数据库名称，比如user_123
    return f"user_{user_info.id}"

async def setup_user_db(token: str = Depends(your_token_auth_scheme)):
    try:
        user_info = parse_token(token)
        db_name = get_user_db_name(user_info)
        # 动态添加数据库连接（如果未配置）
        if db_name not in Tortoise._connections:
            await Tortoise.add_db(
                db_url=f"postgres://username:password@host/{db_name}",
                db_name=db_name
            )
        set_current_db(db_name)
        yield
    finally:
        # 请求结束后清理线程局部变量，避免线程复用导致错误
        if hasattr(THREAD_LOCAL, 'db_name'):
            del THREAD_LOCAL.db_name

然后在路由中使用这个依赖：

@app.get("/user/data")
async def get_user_data(db_setup=Depends(setup_user_db)):
    # 后续数据库操作会自动切换到用户专属库
    ...

3. 自动切换数据库的两种方式 #

方式一：封装模型基类，手动指定数据库 #

定义抽象基类，封装using_current方法，让业务模型继承：

from tortoise.models import Model

class UserSpecificModel(Model):
    class Meta:
        abstract = True

    @classmethod
    def using_current(cls):
        return cls.using(get_current_db())

使用时直接调用using_current()即可操作用户专属库：

# 查询数据
user_data = await UserSpecificModel.using_current().filter(status=1).all()
# 创建数据
await UserSpecificModel.using_current().create(name="test", value=123)

方式二：自定义数据库路由，自动切换 #

通过Tortoise的Router类实现自动路由，无需手动指定数据库：

from tortoise.router import Router

class UserDBRouter(Router):
    def db_for_read(self, model, **hints):
        # 读操作使用当前线程的数据库
        return get_current_db()

    def db_for_write(self, model, **hints):
        # 写操作使用当前线程的数据库
        return get_current_db()

初始化Tortoise时注册这个路由：

await Tortoise.init(
    db_urls={
        "default": "postgres://username:password@host/default_db",  # 公共库
        # 可以提前配置已知用户库，也可以动态添加
    },
    modules={"models": ["your_app.models"]},
    routers=[UserDBRouter()]
)

这样所有模型的读写操作都会自动使用当前线程设置的数据库，无需额外调用using()。

注意事项 #

• 线程清理：必须在请求结束后清理线程局部变量，否则线程池复用会导致后续请求使用错误的数据库。
• 动态数据库：如果用户数据库是动态创建的，要确保添加连接时的异常处理（比如数据库不存在时先创建库）。
• 连接池：注意数据库连接池的配置，避免因用户过多导致连接数耗尽。

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