mysql 读写分离（手动和自动方法）

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

app = Flask(__name__)
app.config[‘SQLALCHEMY_DATABASE_URI‘] = ‘mysql://root:mysql@192.168.105.134:3306/test30‘  # 设置数据库连接地址
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS‘] = False  # 是否追踪数据库变化(触发某些钩子函数), 开启后效率会变
app.config[‘SQLALCHEMY_ECHO‘] = True  # 开启后, 控制台会打印底层执行的SQL语句

app.config[‘SQLALCHEMY_BINDS‘] = {  # get_engine的bind参数为该配置的键
    ‘master‘: ‘mysql://root:mysql@192.168.105.134:3306/test30‘,
    ‘slave‘: ‘mysql://root:mysql@192.168.105.134:8306/test30‘
}


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)


# 初始化组件(建立数据库连接)
db = RoutingSQLAlchemy(app)


# ORM  类->表  类属性->字段  对象->记录
class User(db.Model):
    __tablename__ = "t_user"  # 设置表名, 默认为类名的小写
    id = db.Column(db.Integer, primary_key=True)  # 主键
    name = db.Column(db.String(20), unique=True, nullable=False)  # 设置唯一&非空约束
    age = db.Column(db.Integer, default=10, index=True)  # 设置默认值约束&建立索引


@app.route(‘/‘)
def index():  
    # 增加数据  进行检验是否读写分离
    # 1.创建模型对象
    user1 = User(name=‘zs‘, age=20)
    db.session.add(user1)
    db.session.commit()

    print(‘-‘ * 30)
    # 查询数据  
    print(User.query.all())
    return "index"


if __name__ == ‘__main__‘:
    db.drop_all()  # 删除所有继承自db.Model的表
    db.create_all()  # 创建所有继承自db.Model的表
    app.run(debug=True)

from .routing_sqlalchemy import RoutingSQLAlchemy

db = RoutingSQLAlchemy()

import functools

from user_select import db

def set_read_db(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        db.session().set_to_read()
        return f(*args, **kwargs)
    
    return wrapper


def set_write_db(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        db.session().set_to_write()
        return f(*args, **kwargs)

    return wrapper

from flask import Flask
from user_select import db
from user_select.routing_sqlalchemy import Config
from user_select.decorator import set_read_db, set_write_db


app = Flask(__name__)
app.config.from_object(Config)

# 初始化数据库连接对象
db.init_app(app)


# 建立映射模型  类->表  类属性->字段  对象->记录
class User(db.Model):
   __tablename__ = ‘t_user‘  # 设置表名  表名默认为类名小写
   id = db.Column(db.Integer, primary_key=True)  # 主键  默认主键自增
   name = db.Column(db.String(20), unique=True)  # 设置唯一
   age = db.Column(db.Integer)


@app.route(‘/‘)
@set_write_db
def index():
    """增加数据"""
    user1 = User(name=‘lisi‘, age=20)
    db.session.add(user1)
    db.session.commit()

    return "index"


@app.route(‘/demo1‘)
# @set_read_db
def demo1():
    users = User.query.all()
    print(users)
    return ‘demo1‘


if __name__ == ‘__main__‘:
    # db.drop_all()  # 删除所有继承自db.Model的表
    # db.create_all()  # 创建所有继承自db.Model的表
    app.run(debug=True)