FastAPI多数据库连接实现方法及数据读写操作相关问题求助

核心实现逻辑 #

多数据库连接的核心是为每个数据库单独创建独立的连接引擎、会话实例和模型基类，不同数据库的操作完全隔离，避免会话和引擎绑定混乱。

具体实现步骤 #

1. 依赖安装
   执行命令安装所需工具：
   pip install fastapi sqlalchemy uvicorn pymysql psycopg2-binary
2. 多数据源初始化
   为每个数据库单独配置连接参数、引擎、会话类和模型基类：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 第一个数据库：用户库配置（示例为MySQL）
SQLALCHEMY_DATABASE_URL_USER = "mysql+pymysql://账号:密码@localhost/用户库名"
engine_user = create_engine(SQLALCHEMY_DATABASE_URL_USER)
SessionLocalUser = sessionmaker(autocommit=False, autoflush=False, bind=engine_user)
BaseUser = declarative_base()

# 第二个数据库：订单库配置（示例为PostgreSQL）
SQLALCHEMY_DATABASE_URL_ORDER = "postgresql://账号:密码@localhost/订单库名"
engine_order = create_engine(SQLALCHEMY_DATABASE_URL_ORDER)
SessionLocalOrder = sessionmaker(autocommit=False, autoflush=False, bind=engine_order)
BaseOrder = declarative_base()

注意：不同库的模型类需要绑定对应的基类，比如用户相关模型继承BaseUser，订单相关模型继承BaseOrder
3. 封装会话依赖
为每个数据库单独封装会话获取依赖，实现自动关闭连接：

def get_db_user():
    db = SessionLocalUser()
    try:
        yield db
    finally:
        db.close()

def get_db_order():
    db = SessionLocalOrder()
    try:
        yield db
    finally:
        db.close()

数据库操作示例 #

数据查询 #

在路由函数中通过依赖注入获取对应数据库的会话，即可分别执行查询操作：

from fastapi import FastAPI, Depends
from .models import User, Order

app = FastAPI()

@app.get("/user/{user_id}/orders")
def get_user_orders(user_id: int, db_user = Depends(get_db_user), db_order = Depends(get_db_order)):
    # 从用户库查询用户基础信息
    user = db_user.query(User).filter(User.id == user_id).first()
    # 从订单库查询该用户的所有订单
    orders = db_order.query(Order).filter(Order.user_id == user_id).all()
    return {
        "user_info": user,
        "order_list": orders
    }

数据更新 #

针对需要更新的数据库，使用对应会话执行提交操作即可：

@app.put("/user/{user_id}/balance")
def update_user_balance(user_id: int, increment: int, db_user = Depends(get_db_user)):
    user = db_user.query(User).filter(User.id == user_id).first()
    # 修改字段值
    user.balance += increment
    # 提交更新到数据库
    db_user.add(user)
    db_user.commit()
    # 刷新实例获取最新数据
    db_user.refresh(user)
    return {"current_balance": user.balance}

其他可选实现方案 #

• 异步场景可使用SQLAlchemy异步版本，分别为每个数据库创建AsyncEngine和AsyncSession实例，依赖函数改为异步生成器即可
• 若使用Tortoise-ORM，可直接在初始化配置中定义多个数据源，每个模型通过Meta类的app参数绑定对应数据库
• 轻量无ORM场景可直接使用databases库，为每个数据库创建独立的连接实例，按需调用即可

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