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

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

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] 纯文本查看 复制代码

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()

扫码关注黑马视频库公众号，免费获取黑马全学科视频教程

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

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

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] 纯文本查看 复制代码

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()

扫码关注黑马视频库公众号，免费获取黑马全学科视频教程