使用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