SQLAlchemy分表策略怎么设计

本文目录导读：

1. 基础分表策略：使用 __tablename__ 动态生成
2. 动态创建模型类（推荐）
3. 使用 SQLAlchemy 的 baked 查询 + 路由
4. 高级方案：使用 __init_subclass__ 或元类
5. 与数据库的自动分表结合
6. 推荐的分表策略选择
7. 注意事项

SQLAlchemy 的分表（分片/Sharding）通常需要结合业务场景来设计，常见策略包括 水平分表（按时间、按ID范围、按哈希）和 垂直分表（按字段拆分）,以下是几种主流方案的详细设计：

基础分表策略：使用 __tablename__ 动态生成

最直接的方式是继承基类并动态设置表名。

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
Base = declarative_base()
class Order(Base):
    __abstract__ = True  # 不创建真实的表
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer)
    amount = Column(Integer)
    created_at = Column(DateTime)
class Order202401(Order):
    __tablename__ = 'orders_202401'
class Order202402(Order):
    __tablename__ = 'orders_202402'

优点：简单直观
缺点：表数量需要手动维护，不适合动态分表

动态创建模型类（推荐）

通过工厂函数动态生成模型类，适合按时间、ID范围分表。

import datetime
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
def get_order_table(year, month):
    table_name = f'orders_{year}{month:02d}'
    # 检查是否已经有该表名的类
    for cls in Base.registry._class_registry.values():
        if hasattr(cls, '__tablename__') and cls.__tablename__ == table_name:
            return cls
    # 动态创建新类
    class DynamicOrder(Base):
        __tablename__ = table_name
        __table_args__ = {'extend_existing': True}  # 允许覆盖
        id = Column(Integer, primary_key=True)
        user_id = Column(Integer)
        amount = Column(Integer)
        created_at = Column(DateTime, default=datetime.datetime.now)
    return DynamicOrder
# 使用示例
now = datetime.datetime.now()
OrderModel = get_order_table(now.year, now.month)
session.add(OrderModel(user_id=123, amount=500))

使用 SQLAlchemy 的 baked 查询 + 路由

适用于需要根据某个字段自动路由到不同表的场景。

from sqlalchemy.orm import Session
from sqlalchemy import text
class ShardRouter:
    def __init__(self, shard_key_field='user_id'):
        self.shard_key_field = shard_key_field
        self.shards = 10  # 分10张表
    def get_table_name(self, shard_key):
        shard_id = shard_key % self.shards
        return f'users_{shard_id}'
    def add_user(self, session, user_data):
        table_name = self.get_table_name(user_data['id'])
        # 动态执行INSERT
        session.execute(
            text(f"INSERT INTO {table_name} (id, name) VALUES (:id, :name)"),
            user_data
        )
    def get_user(self, session, user_id):
        table_name = self.get_table_name(user_id)
        result = session.execute(
            text(f"SELECT * FROM {table_name} WHERE id = :id"),
            {'id': user_id}
        )
        return result.fetchone()
# 使用
router = ShardRouter()
router.add_user(session, {'id': 123, 'name': 'Alice'})
user = router.get_user(session, 123)

高级方案：使用 __init_subclass__ 或元类

适用于需要自动注册所有分表类的场景。

class ShardMeta(type(Base)):
    _shard_registry = {}
    def __new__(cls, name, bases, attrs):
        new_cls = super().__new__(cls, name, bases, attrs)
        if 'shard_key' in attrs:
            cls._shard_registry[attrs['shard_key']] = new_cls
        return new_cls
class BaseShard(metaclass=ShardMeta):
    __abstract__ = True
    id = Column(Integer, primary_key=True)
    @classmethod
    def get_shard(cls, shard_key):
        return cls._shard_registry[shard_key]
# 自动注册
class UserShard0(BaseShard):
    __tablename__ = 'users_0'
    shard_key = 0
class UserShard1(BaseShard):
    __tablename__ = 'users_1'
    shard_key = 1

与数据库的自动分表结合

如果使用 PostgreSQL 的分区表，SQLAlchemy 可以直接操作：

# 创建分区表（在数据库中）
"""
CREATE TABLE orders (
    id SERIAL,
    user_id INT,
    amount DECIMAL,
    created_at TIMESTAMP
) PARTITION BY RANGE (created_at);
CREATE TABLE orders_2024_q1 PARTITION OF orders
    FOR VALUES FROM ('2024-01-01') TO ('2024-04-01');
"""
class Order(Base):
    __tablename__ = 'orders'
    __table_args__ = {'extend_existing': True}
    id = Column(Integer, primary_key=True)
    # ... 其他字段
# 直接操作主表，数据库自动路由到分区
session.add(Order(created_at=datetime(2024, 2, 1)))

推荐的分表策略选择

注意事项

1. 事务一致性：跨表操作需要分布式事务框架（如 saga 模式）
2. 查询路由：对不支持分表键的查询要降级为全表扫描（需限制）
3. 表创建：使用 Base.metadata.create_all() 不会自动创建动态表，需要手动或使用事件监听
4. 模型注册：动态创建的类需要确保不重复注册（使用 extend_existing 或缓存）
5. 迁移工具：Alembic 对动态分表支持有限，建议手动管理分表DDL

如果需要更复杂的分布式事务支持，可以结合 sqlalchemy-sharding 扩展或使用分布式数据库中间件（如 MyCat、ShardingSphere）。