SQLAlchemy多绑定场景下为同结构多诊所数据库动态选择查询绑定

实现方案代码 #

以下代码基于Flask-SQLAlchemy实现，兼容2.x/3.x版本，完全满足你提出的两个需求：

import contextvars
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, Integer, String, Date

# 1. 基础配置初始化
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:pass@localhost/main'
app.config['SQLALCHEMY_BINDS'] = {
    'clinic1':'mysql://user:pass@localhost/clinic1',
    'clinic2':'mysql://user:pass@localhost/clinic2',
    'clinic3':'mysql://user:pass@localhost/clinic3',
    'clinic4':'mysql://user:pass@localhost/clinic4'
}
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 2. 定义上下文变量，存储当前使用的数据库绑定
current_bind = contextvars.ContextVar('current_bind', default=None)

# 3. 实现数据库绑定上下文管理器
class DbContext:
    def __init__(self, bind):
        if bind not in app.config['SQLALCHEMY_BINDS']:
            raise ValueError(f"无效的数据库绑定: {bind}")
        self.bind = bind
        self.token = None

    def __enter__(self):
        self.token = current_bind.set(self.bind)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        current_bind.reset(self.token)
        return False

# 4. 定义诊所业务模型抽象基类，所有业务表继承该类即可
class ClinicBaseModel(db.Model):
    __abstract__ = True

    @property
    def __bind_key__(self):
        bind = current_bind.get()
        if bind is None:
            raise RuntimeError("请在DbContext上下文中执行诊所业务表操作，未指定绑定数据库")
        return bind

# 5. 定义具体业务模型
class Patient(ClinicBaseModel):
    __tablename__ = "patients"
    id = Column(Integer, primary_key=True)
    first_name = Column(String, index=True)
    last_name = Column(String, index=True)
    date_of_birth = Column(Date, index=True)

# Doctor、Appointment等其他业务模型同理，继承ClinicBaseModel即可
# class Doctor(ClinicBaseModel):
#     __tablename__ = "doctors"
#     ...

# 6. 实现多库批量建表方法
def create_clinic_tables():
    # 遍历所有诊所绑定，逐个建表
    for bind_key in app.config['SQLALCHEMY_BINDS'].keys():
        engine = db.get_engine(app, bind=bind_key)
        ClinicBaseModel.metadata.create_all(engine)

使用示例 #

with app.app_context():
    # 建表：执行后clinic1-clinic4四个库都会自动创建patients等业务表
    create_clinic_tables()

    # 1. 上下文内查询：自动路由到指定绑定库
    with DbContext(bind='clinic1'):
        patients_count = Patient.query.filter().count()
        print(f"clinic1患者数量：{patients_count}")

        # 写入操作同样自动路由到对应库
        new_patient = Patient(first_name='张', last_name='三', date_of_birth='1990-01-01')
        db.session.add(new_patient)
        db.session.commit()

    # 2. 无上下文直接操作：自动抛出错误，符合要求
    try:
        Patient.query.filter().count()
    except RuntimeError as e:
        print(f"预期报错：{e}")

实现说明 #

• 基于contextvars实现绑定状态存储，线程/协程安全，多请求场景下不会出现绑定串扰
• 所有业务模型统一继承抽象基类，无需重复实现绑定逻辑，新增业务表只需继承ClinicBaseModel即可
• 建表逻辑仅针对配置的诊所绑定库执行，默认main库不会创建任何业务表

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