這篇文章主要介紹了Python實現自定義讀寫分離程式碼例項,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

思路

• 自定義Session類
  • 重寫get_bind方法
  • 根據self._flushing判斷讀寫操作,選擇對應的資料庫
• 自定義SQLAlchemy類
  • 重寫create_session,在其中使用自定義的Session類

from flask import Flask
from flask_sqlalchemy import SQLAlchemy,SignallingSession,get_state
from sqlalchemy import orm


class RoutingSession(SignallingSession):
  def get_bind(self,mapper=None,clause=None):
    state = get_state(self.app)

    # 判斷讀寫操作
    if self._flushing: # 寫操作,使用主資料庫
      print("寫入資料")
      return state.db.get_engine(self.app,bind='master')
    else: # 讀操作,使用從資料庫
      print('讀取資料')
      return state.db.get_engine(self.app,bind='slave')


class RoutingSQLAlchemy(SQLAlchemy):
  def create_session(self,options):
    return orm.sessionmaker(class_=RoutingSession,db=self,**options)


app = Flask(__name__)
# 設定資料庫的連線地址
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:[email protected]:3306/demo'
# 設定資料庫的繫結地址
app.config['SQLALCHEMY_BINDS'] = {
  'master': "mysql://root:[email protected]:3306/demo",'slave': "mysql://root:[email protected]:8306/demo"
}
# 設定是否追蹤資料庫變化  一般不會開啟,影響效能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 設定是否列印底層執行的SQL語句
app.config['SQLALCHEMY_ECHO'] = False

# 建立資料庫連線物件
db = RoutingSQLAlchemy(app)

# 使用者表 一
class User(db.Model):
  __tablename__ = 't_user'
  id = db.Column(db.Integer,primary_key=True)
  name = db.Column(db.String(20),unique=True)


@app.route('/')
def index():
  # 增加資料
  user1 = User(name='zs')
  db.session.add(user1)
  db.session.commit()

  # 查詢資料
  users = User.query.all()
  print(users)
  return "index"

if __name__ == '__main__':
  # 刪除所有繼承自db.Model的表
  db.drop_all()
  # 建立所有繼承自db.Model的表
  db.create_all()
  app.run(debug=True)

不太好，自動選擇不能控制

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。