SQLAlchemy分表策略怎么设计

wen python案例 2026-07-12 1

本文目录导读：

SQLAlchemy分表策略怎么设计

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

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

业务场景	推荐策略	优点
按月分表	动态模型工厂	灵活自动创建
按用户ID分片	哈希路由 + 原生SQL	数据分布均匀
固定数量的分表	预定义多个模型类	简单稳定
使用数据库分区	PostgreSQL/MySQL分区	对应用透明
需要ORM完整支持	动态模型 + `with_polymorphic`	支持关系映射