使用flask-sqlalchemy 实现mysql读写分离

在咱们的学习过程当中，django项目给你们详细的解释了如何配置主从读写分离，可是flask中没有提到相关的知识，这里小编为你们搜集了一篇简单的利用flask-SQLAlchemy为你们搭建mysql的主从读写分离。
flask中没有直接的主从读写分离配置，须要本身创造相应的引擎，构造读模式和写模式的隐射关系，经过判断是读操做或者写操做，链接对应的数据库。

[Python]  纯文本查看 复制代码

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker   # 建立对象的基类,全部model继承base Base = declarative_base() def get_db_sessionmaker(db_uri, * * kvargs):      # 初始化数据库链接:      engine = create_engine(db_uri, * * kvargs)      # 建立DBSession类型:      DBSession = sessionmaker(bind = engine)      DBSession.configure(bind = engine)      Base.metadata.create_all(engine)      return DBSession

[Python]  纯文本查看 复制代码

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48

import threading from contextlib import contextmanager from frame.orm.db_util import get_db_sessionmaker, Base from utils.collection_util import is_empty from utils.random_util import randchoice   DEFAULT_SESSION_NAME = "default" Base = Base class DBManager( object ):      singleton = None      mutex = threading.Lock()        def __init__( self ):             self .__conf = {}          self .__session_dict = {}       def db_uri( self , name, db_uri, * * kwargs):          self .__conf[name] = db_uri          self .__session_dict[name] = get_db_sessionmaker(db_uri, * * kwargs)          return self        @staticmethod      def get_instance():             if DBManager.singleton = = None :              DBManager.mutex.acquire()              if DBManager.singleton = = None :                  DBManager.singleton = DBManager()              DBManager.mutex.release()          return DBManager.singleton        def __get_session_maker( self , name = None ):          if is_empty( self .__session_dict) or is_empty( self .__conf):              raise Exception( "__session_dict is null,please add dburi to dbmanager" )          if name = = None :              name = randchoice( self .__session_dict.keys())          return self .__session_dict[name]        @contextmanager      def session_ctx( self , name = None ):          DBSession = self .__get_session_maker(name)          session = DBSession()          try :              yield session              session.commit()          except Exception, ex:              session.rollback()          finally :              session.close()

更多技术资讯可关注：gzitcast