如何在FastAPI项目中使用SQLModel连接多个数据库？

核心逻辑就是给每个数据库单独配置一套元数据、引擎和会话，让它们各自独立，互不干扰。下面是具体实现步骤：

1. 配置多个数据库连接地址 #

直接定义不同数据库的URL即可，比如SQLite和PostgreSQL混用：

from sqlmodel import SQLModel, create_engine
from fastapi import Depends
from sqlalchemy.orm import sessionmaker

# 数据库1（SQLite）
DB1_URL = "sqlite:///./db1.db"
# 数据库2（PostgreSQL）
DB2_URL = "postgresql://user:password@localhost/db2"

2. 为每个数据库创建独立的引擎和会话工厂 #

每个数据库单独生成引擎，会话工厂绑定对应引擎，避免混用：

# 数据库1的引擎与会话
engine_db1 = create_engine(DB1_URL, echo=True)
SessionLocalDB1 = sessionmaker(autocommit=False, autoflush=False, bind=engine_db1)

# 数据库2的引擎与会话
engine_db2 = create_engine(DB2_URL, echo=True)
SessionLocalDB2 = sessionmaker(autocommit=False, autoflush=False, bind=engine_db2)

3. 给每个数据库的模型绑定专属元数据 #

这是关键！默认SQLModel会用全局元数据，必须为每个数据库自定义元数据，然后在模型里指定metadata参数，明确模型归属：

from sqlalchemy import MetaData

# 分别创建两个数据库的元数据对象
metadata_db1 = MetaData()
metadata_db2 = MetaData()

# 数据库1的模型，绑定metadata_db1
class UserDB1(SQLModel, table=True, metadata=metadata_db1):
    id: int | None = SQLModel.Field(default=None, primary_key=True)
    name: str

# 数据库2的模型，绑定metadata_db2
class ProductDB2(SQLModel, table=True, metadata=metadata_db2):
    id: int | None = SQLModel.Field(default=None, primary_key=True)
    product_name: str

4. 初始化数据库表（可选） #

首次运行时，需要为每个数据库创建表，用各自的元数据绑定引擎执行：

# 创建数据库1的所有表
metadata_db1.create_all(bind=engine_db1)
# 创建数据库2的所有表
metadata_db2.create_all(bind=engine_db2)

5. 定义每个数据库的会话依赖 #

为每个数据库写独立的依赖函数，供FastAPI端点注入使用：

def get_db1():
    db = SessionLocalDB1()
    try:
        yield db
    finally:
        db.close()

def get_db2():
    db = SessionLocalDB2()
    try:
        yield db
    finally:
        db.close()

6. 在端点中使用对应数据库 #

直接在端点参数里注入对应依赖，操作时用注入的会话即可：

from fastapi import FastAPI

app = FastAPI()

# 该端点使用数据库1查询用户
@app.get("/users")
def get_users(db=Depends(get_db1)):
    users = db.query(UserDB1).all()
    return users

# 该端点使用数据库2查询商品
@app.get("/products")
def get_products(db=Depends(get_db2)):
    products = db.query(ProductDB2).all()
    return products

额外提醒 #

• 模型的元数据必须和对应数据库引擎绑定，否则会出现表找不到或跨库创建的问题
• 如果使用异步SQLModel，只需把create_engine换成create_async_engine，会话工厂用async_sessionmaker，依赖函数改为异步即可，逻辑完全一致

内容的提问来源于stack exchange，提问作者configure.ng