SQLAlchemy读写分离怎么实现

wen python案例 2026-07-12 184

本文目录导读：

SQLAlchemy读写分离怎么实现

使用内置的HorizontallyShard（推荐）
使用SQLAlchemy的Events方式
使用自定义Session实现
完整的生产环境实现
使用第三方库（简单方案）
注意事项

在SQLAlchemy中实现读写分离主要有以下几种方式,我来详细介绍：

使用内置的HorizontallyShard（推荐）

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.ext.horizontal_shard import ShardedSession
# 配置主从数据库
engines = {
    'master': create_engine('mysql://user:pass@master_host/dbname'),
    'slave1': create_engine('mysql://user:pass@slave1_host/dbname'),
    'slave2': create_engine('mysql://user:pass@slave2_host/dbname'),
}
class RoutingSession(Session):
    # 获取查询使用的数据库（读操作）
    def get_bind(self, mapper=None, clause=None, **kwargs):
        if self._flushing or (mapper and self._is_write_operation(mapper)):
            return engines['master']
        else:
            # 随机选择从库
            import random
            return random.choice([engines['slave1'], engines['slave2']])
    def _is_write_operation(self, mapper):
        # 判断当前操作是否为写操作
        return self.is_modified() or self.deleted
# 创建Session工厂
Session = sessionmaker(class_=RoutingSession)

使用SQLAlchemy的Events方式

from sqlalchemy import event, create_engine
from sqlalchemy.orm import Session, sessionmaker
class RoutingSession(Session):
    def __init__(self, *args, **kwargs):
        self._write_engine = kwargs.pop('write_engine', None)
        self._read_engines = kwargs.pop('read_engines', [])
        super().__init__(*args, **kwargs)
    def get_bind(self, mapper=None, clause=None, **kwargs):
        # 获取当前是否处于写操作状态
        if self.info.get('write'):
            return self._write_engine
        else:
            import random
            return self._read_engines[random.randint(0, len(self._read_engines)-1)]
# 创建数据库引擎
write_engine = create_engine('mysql://user:pass@master_host/dbname')
read_engines = [
    create_engine('mysql://user:pass@slave1_host/dbname'),
    create_engine('mysql://user:pass@slave2_host/dbname'),
]
# 绑定事件监听
@event.listens_for(RoutingSession, 'before_flush')
def before_flush(session, flush_context, instances):
    session.info['write'] = True
@event.listens_for(RoutingSession, 'after_flush')
def after_flush(session, flush_context):
    session.info['write'] = False
# 创建Session
session_factory = sessionmaker(
    class_=RoutingSession,
    write_engine=write_engine,
    read_engines=read_engines
)

使用自定义Session实现

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm.session import SessionTransaction
import logging
logger = logging.getLogger(__name__)
class ReadWriteSession(Session):
    def __init__(self, *args, **kwargs):
        self.read_engine = kwargs.pop('read_engine', None)
        self.write_engine = kwargs.pop('write_engine', None)
        super().__init__(*args, **kwargs)
    def get_bind(self, mapper=None, clause=None, **kw):
        # 获取当前绑定的数据库
        if hasattr(self, 'transaction') and self.transaction is not None:
            # 如果在事务中，使用写库
            return self.write_engine
        # 检查是否写操作
        if self.info.get('operating') == 'write':
            return self.write_engine
        return self.read_engine
# 配置数据库
read_engine = create_engine('mysql://user:pass@slave_host/dbname')
write_engine = create_engine('mysql://user:pass@master_host/dbname')
# 创建Session工厂
Session = sessionmaker(
    class_=ReadWriteSession,
    read_engine=read_engine,
    write_engine=write_engine,
    expire_on_commit=False
)
# 使用示例
def create_user(db_session, user_data):
    # 手动标记为写操作
    db_session.info['operating'] = 'write'
    user = User(**user_data)
    db_session.add(user)
    db_session.commit()
    db_session.info['operating'] = None
    return user
def get_users(db_session):
    # 读操作，默认使用从库
    return db_session.query(User).all()

完整的生产环境实现

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm.session import SessionTransaction
from contextlib import contextmanager
from typing import List, Optional
import random
class DatabaseRouter:
    """数据库读写分离路由"""
    def __init__(self, master_url: str, slave_urls: List[str], echo=False):
        self.master_engine = create_engine(master_url, echo=echo)
        self.slave_engines = [
            create_engine(url, echo=echo) for url in slave_urls
        ]
        self._session_factory = None
    @property
    def session_factory(self):
        if self._session_factory is None:
            self._session_factory = sessionmaker(
                class_=_ReadWriteSession,
                write_engine=self.master_engine,
                read_engines=self.slave_engines
            )
        return self._session_factory
    @contextmanager
    def get_session(self, force_write=False):
        """获取数据库会话"""
        session = self.session_factory()
        try:
            if force_write:
                session.info['write'] = True
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
    @contextmanager
    def write_session(self):
        """获取写会话（自动标记为写操作）"""
        with self.get_session(force_write=True) as session:
            yield session
class _ReadWriteSession(Session):
    def __init__(self, *args, **kwargs):
        self.write_engine = kwargs.pop('write_engine', None)
        self.read_engines = kwargs.pop('read_engines', [])
        super().__init__(*args, **kwargs)
    def get_bind(self, mapper=None, clause=None, **kw):
        # 写操作或事务中使用主库
        if self.info.get('write', False) or self._is_in_transaction():
            return self.write_engine
        # 读操作随机选择从库
        return random.choice(self.read_engines)
    def _is_in_transaction(self):
        """检查是否在事务中"""
        try:
            return self.transaction is not None
        except Exception:
            return False
# 使用示例
router = DatabaseRouter(
    master_url='mysql://user:pass@master_host:3306/dbname',
    slave_urls=[
        'mysql://user:pass@slave1_host:3306/dbname',
        'mysql://user:pass@slave2_host:3306/dbname',
    ],
    echo=False
)
# 写操作
with router.write_session() as session:
    user = User(name='Alice', email='alice@example.com')
    session.add(user)
# 读操作（自动使用从库）
with router.get_session() as session:
    users = session.query(User).all()

使用第三方库（简单方案）

# 安装：pip install sqlalchemy-readwrite
from sqlalchemy_readwrite.middleware import ReadWriteMiddleware
# 配置
middleware = ReadWriteMiddleware(
    write_db_url='mysql://user:pass@master_host/dbname',
    read_db_urls=[
        'mysql://user:pass@slave1_host/dbname',
        'mysql://user:pass@slave2_host/dbname',
    ]
)
# 获取session
session = middleware.get_session()