FastAPI实现多模型连接不同数据库及解决循环导入问题

问题背景 #

需要在FastAPI中实现类多租户应用，让部分模型连接与其他模型不同的数据库，但尝试多种方案时始终遭遇循环导入问题。

文件结构 #

app
 |- database/
    |- models/
       |- api_tokens.py（仅包含ApiToken类）
       |- users.py（仅包含User类）
       |- 其他模型文件
    |- base.py
    |- session.py
 |- routers/
    |- 各路由文件独立存放
 |- main.py

初始文件内容 #

app/database/session.py

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Generator

SQLALCHEMY_DATABASE_URI = "mysql+pymysql://" + os.getenv('DB_USERNAME') + ":" + os.getenv('DB_PASSWORD') + "@" + os.getenv('DB_HOST') + "/" + os.getenv('DB_DATABASE') + "?charset=utf8mb4"
SQLALCHEMY_CUSTOMERS_DATABASE_URI = "mysql+pymysql://" + os.getenv('CUSTOMERS_DB_USERNAME') + ":" + os.getenv('CUSTOMERS_DB_PASSWORD') + "@" + os.getenv('CUSTOMERS_DB_HOST') + "/" + os.getenv('CUSTOMERS_DB_DATABASE') + "?charset=utf8mb4"

engine = create_engine(SQLALCHEMY_DATABASE_URI)
customers_engine = create_engine(SQLALCHEMY_CUSTOMERS_DATABASE_URI)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind = engine)

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

app/database/base.py

from sqlalchemy.orm import DeclarativeBase, Query
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

from .session import get_db

class Base(DeclarativeBase):
    session = next(get_db())

    id: Mapped[str] = mapped_column(primary_key = True)

class ApiTokensBase(DeclarativeBase):
    session = next(get_db())

    id: Mapped[str] = mapped_column(primary_key = True)

核心问题 #

如何让继承ApiTokensBase类的ApiToken模型使用SQLALCHEMY_CUSTOMERS_DATABASE_URI指定的数据库连接？

已尝试方案 #

尝试通过SQLAlchemy的session binds特性，将不同基类绑定到对应数据库引擎，但在session.py中导入base.py的Base和ApiTokensBase时，出现循环导入问题。尝试的session.py代码如下：

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Generator

from app.database.base import Base, ApiTokensBase

SQLALCHEMY_DATABASE_URI = "mysql+pymysql://" + os.getenv('DB_USERNAME') + ":" + os.getenv('DB_PASSWORD') + "@" + os.getenv('DB_HOST') + "/" + os.getenv('DB_DATABASE') + "?charset=utf8mb4"
SQLALCHEMY_CUSTOMERS_DATABASE_URI = "mysql+pymysql://" + os.getenv('CUSTOMERS_DB_USERNAME') + ":" + os.getenv('CUSTOMERS_DB_PASSWORD') + "@" + os.getenv('CUSTOMERS_DB_HOST') + "/" + os.getenv('CUSTOMERS_DB_DATABASE') + "?charset=utf8mb4"

engine = create_engine(SQLALCHEMY_DATABASE_URI)
customers_engine = create_engine(SQLALCHEMY_CUSTOMERS_DATABASE_URI)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind = engine, binds = {
    Base: engine,
    ApiTokensBase: customers_engine,
})

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

解决方案 #

1. 修正base.py，打破循环导入依赖 #

原base.py中错误地在基类中绑定了全局session实例，同时导入了session.py的get_db，这是循环导入的根源。修改后：

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    id: Mapped[str] = mapped_column(primary_key=True)

class ApiTokensBase(DeclarativeBase):
    id: Mapped[str] = mapped_column(primary_key=True)

• 移除对session.py的导入，基类仅负责定义模型结构，不持有任何session实例。

2. 修正session.py，正确配置多引擎绑定 #

现在session.py可以安全导入基类，配置sessionmaker时通过binds指定不同基类对应的引擎：

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Generator

# 安全导入基类，无循环依赖
from app.database.base import Base, ApiTokensBase

# 简化环境变量拼接
SQLALCHEMY_DATABASE_URI = (
    f"mysql+pymysql://{os.getenv('DB_USERNAME')}:{os.getenv('DB_PASSWORD')}@"
    f"{os.getenv('DB_HOST')}/{os.getenv('DB_DATABASE')}?charset=utf8mb4"
)
SQLALCHEMY_CUSTOMERS_DATABASE_URI = (
    f"mysql+pymysql://{os.getenv('CUSTOMERS_DB_USERNAME')}:{os.getenv('CUSTOMERS_DB_PASSWORD')}@"
    f"{os.getenv('CUSTOMERS_DB_HOST')}/{os.getenv('CUSTOMERS_DB_DATABASE')}?charset=utf8mb4"
)

engine = create_engine(SQLALCHEMY_DATABASE_URI)
customers_engine = create_engine(SQLALCHEMY_CUSTOMERS_DATABASE_URI)

# 配置session，绑定不同基类到对应引擎
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
    binds={
        Base: engine,
        ApiTokensBase: customers_engine
    }
)

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

3. 模型文件示例（api_tokens.py） #

确保ApiToken继承ApiTokensBase，自动关联到客户数据库：

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from app.database.base import ApiTokensBase

class ApiToken(ApiTokensBase):
    __tablename__ = "api_tokens"
    token: Mapped[str] = mapped_column(unique=True, nullable=False)
    # 其他业务字段...

4. 路由中使用session #

在路由中依旧通过get_db获取session，SQLAlchemy会自动根据模型的基类选择对应的数据库引擎：

from fastapi import Depends, APIRouter
from sqlalchemy.orm import Session
from app.database.session import get_db
from app.database.models.api_tokens import ApiToken

router = APIRouter()

@router.get("/tokens/{token_id}")
def get_token(token_id: str, db: Session = Depends(get_db)):
    return db.query(ApiToken).filter(ApiToken.id == token_id).first()

关键说明 #

• 循环导入的核心原因是base.py依赖session.py，同时session.py需要导入base.py的基类，修正后打破了这个依赖链。
• SQLAlchemy的binds特性会根据模型的DeclarativeBase子类自动路由到对应的数据库引擎，无需手动切换连接。
• session是请求级别的资源，不应绑定在模型基类上，需通过get_db生成并注入到路由中。

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