动态切换数据库链接
如果你有多个数据库链接，并且在启动服务之前没有写入相关的配置文件，这种情况下该如何使用动态的增加数据库链接并使用呢？
前置条件
这里假设你的数据库配置信息已经写入默认的数据库中（这里暂且称为应用程序数据库，或者AppDB），在AppDB 中的某张表中存储着其他的数据库链接信息（这里暂且称为应用程序扩展数据库，或者AppExtendDB）
Flask
使用 Flask-SQLAlchemy 创建 model
# 使用 Flask-SQLAlchemy 创建 model from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)email = db.Column(db.String(120), unique=True, nullable=False)# 查询所有用户，使用 AppDB 的数据库链接
User.query.all()def new_session(url):engine = create_engine(url)# 创建一个新的会话Session = sessionmaker(bind=engine) return Session# 创建一个 AppExtendDB 的数据库链接
db_new_session= new_session(url)()
# 查询 AppExtendDB 中的 user 信息，其他诸如表/列必须存在的基本要求不做赘述
User.query.with_session(bb).all()# 类似的只有 AppExtendDB 中存在 UserExtend
# 所以 UserExtend.query.all() 就会报错
# 必须使用 UserExtend.query.with_session(bb).all()
class UserExtend(db.Model):id = db.Column(db.Integer, primary_key=True)age= db.Column(db.String(80), unique=True, nullable=False)phone= db.Column(db.String(120), unique=True, nullable=False)Django
使用 DatabaseRouter
- 创建一个数据库路由类，用于根据需要切换数据库连接。在你的应用程序中创建一个名为 database_router.py 的文件，并添加以下内容：
class DatabaseRouter:def db_for_read(self, model, **hints):# 根据需要返回读取操作的数据库别名if model._meta.app_label == 'AppDB':return 'AppDB url'if model._meta.app_label == 'AppExtendDB':return 'AppExtendDB url'return Nonedef db_for_write(self, model, **hints):# 根据需要返回读取操作的数据库别名if model._meta.app_label == 'AppDB':return 'AppDB url'if model._meta.app_label == 'AppExtendDB':return 'AppExtendDB url'return Nonedef allow_relation(self, obj1, obj2, **hints):# 允许关系操作return Truedef allow_migrate(self, db, app_label, model_name=None, **hints):# 允许迁移操作return True
- 在 settings.py 文件中配置数据库路由：
DATABASE_ROUTERS = ['your_app.database_router.DatabaseRouter']
- 修改你的代码，比如在中间件中
def process_view(request, view_func, view_args, view_kwargs):default_dict = copy.deepcopy(settings.DATABASES['default'])default_dict.update({'ENGINE': 'django.db.backends.postgresql','NAME': 'db name','USER': 'db user','PASSWORD': 'db pwd','HOST': 'db host','PORT': '5432',})settings.DATABASES['AppExtendDB'] = default_dict
- 查询
# 方式一
UserExtend._meta.app_label = "AppExtendDB"
UserExtend.objects.all()# 方式二
UserExtend.objects.using('AppExtendDB').all()到此结 DragonFangQy 2023.10.31
