動(dòng)態(tài)切換數(shù)據(jù)庫(kù)鏈接
如果你有多個(gè)數(shù)據(jù)庫(kù)鏈接籍嘹，并且在啟動(dòng)服務(wù)之前沒(méi)有寫入相關(guān)的配置文件奴璃，這種情況下該如何使用動(dòng)態(tài)的增加數(shù)據(jù)庫(kù)鏈接并使用呢？
前置條件
這里假設(shè)你的數(shù)據(jù)庫(kù)配置信息已經(jīng)寫入默認(rèn)的數(shù)據(jù)庫(kù)中（這里暫且稱為應(yīng)用程序數(shù)據(jù)庫(kù)居暖，或者AppDB）米诉，在AppDB 中的某張表中存儲(chǔ)著其他的數(shù)據(jù)庫(kù)鏈接信息（這里暫且稱為應(yīng)用程序擴(kuò)展數(shù)據(jù)庫(kù)，或者AppExtendDB）
Flask
使用 Flask-SQLAlchemy 創(chuàng)建 model
# 使用 Flask-SQLAlchemy 創(chuàng)建 model
from flask_sqlalchemy import SQLAlchemy
db = 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 的數(shù)據(jù)庫(kù)鏈接
User.query.all()
def new_session(url):
engine = create_engine(url)
# 創(chuàng)建一個(gè)新的會(huì)話
Session = sessionmaker(bind=engine)
return Session
# 創(chuàng)建一個(gè) AppExtendDB 的數(shù)據(jù)庫(kù)鏈接
db_new_session= new_session(url)()
# 查詢 AppExtendDB 中的 user 信息砾层，其他諸如表/列必須存在的基本要求不做贅述
User.query.with_session(bb).all()
# 類似的只有 AppExtendDB 中存在 UserExtend
# 所以 UserExtend.query.all() 就會(huì)報(bào)錯(cuò)
# 必須使用 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
- 創(chuàng)建一個(gè)數(shù)據(jù)庫(kù)路由類，用于根據(jù)需要切換數(shù)據(jù)庫(kù)連接贱案。在你的應(yīng)用程序中創(chuàng)建一個(gè)名為 database_router.py 的文件肛炮，并添加以下內(nèi)容：
class DatabaseRouter:
def db_for_read(self, model, **hints):
# 根據(jù)需要返回讀取操作的數(shù)據(jù)庫(kù)別名
if model._meta.app_label == 'AppDB':
return 'AppDB url'
if model._meta.app_label == 'AppExtendDB':
return 'AppExtendDB url'
return None
def db_for_write(self, model, **hints):
# 根據(jù)需要返回讀取操作的數(shù)據(jù)庫(kù)別名
if model._meta.app_label == 'AppDB':
return 'AppDB url'
if model._meta.app_label == 'AppExtendDB':
return 'AppExtendDB url'
return None
def allow_relation(self, obj1, obj2, **hints):
# 允許關(guān)系操作
return True
def allow_migrate(self, db, app_label, model_name=None, **hints):
# 允許遷移操作
return True
- 在 settings.py 文件中配置數(shù)據(jù)庫(kù)路由：
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
- 查詢
# 方式一 必須和 DatabaseRouter 配合使用
UserExtend._meta.app_label = "AppExtendDB"
UserExtend.objects.all()
# 方式二
UserExtend.objects.using('AppExtendDB').all()
到此結(jié)? DragonFangQy 2023.10.31
