1.需求及场景概述
现有系统中因历史数据量过大，产生了将历史数据进行按月存储的要求，系统和数据库交互使用的是sqlalchemy，假设系统的原来的历史记录表（record）如下：
为了将历史数据按月分表存储，我们需要以此表为基础按月创建对应的月表来进行分表存储，同时又要使用orm的功能。面对这样的需求我们很自然的会想到创建如下模型
class Recode_202405(Base): __tablename__ = 'record_202405' id = Column(INT(11), primary_key=True) name = Column(String(100, 'utf8mb4_unicode_ci'))这样当然可以，但是我们不可能每月手动去创建这个模型，然后重启自己的服务，这明显有问题， 那如何解决呢，下面就介绍一种在这种场景下基于sqlalchemy实现的分表存储方案。
首先，我们对我们应用场景及需求进行一下描述：
1.我们有一张基础表，这个表作为我们创建月表的模板。
2.当有新数据需要入库时，我们将数据按月存储到当前月份的月表中，如果当月表不存在系统自动创建。
3.支持使用OMR方便数据存储及查询。
2.实现方案
既然要支持使用ORM ，我们势必要获取月表的model。sqlalchemy0.9.1版本推出了Automap，它可以自动映射数据库的表，通过数据表名映射model。大概的用法如下：
from sqlalchemy.ext.automap import automap_baseAutoBase = automap_base()# reflect the tables
AutoBase.prepare(engine, reflect=True)
tablename = "record_202405"
RecordDao = getattr(AutoBase.classes, tablename)既然可以通过表名映射回model,那么现在的问题就是如何基于已有的基础表创建当月的月表，可以参考如下代码：
from sqlalchemy.ext.automap import automap_base
AutoBase = automap_base()table_name = 'record'
date = '202405'
# 基于基础表创建指定月份的月表
base_table = autobase.metadata.tables[table_name]
mdata = MetaData()
new_table = base_table.tometadata(metadata=mdata, name=f'{table_name}_{date}')
try:# 创建表new_table.create(bind=engine)
except BaseException as e:print(e)基于上述方案，我们可以封装一个方法，该方法接收两个参数，一个是表名，一个是分表的月份，其实可以进行任何程度的分表，返回该表的model,通过sqlalchemy 的ORM 进行增删改查操作，整体代码如下：
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.automap import automap_base
from urllib import parse
from contextlib import contextmanager
import timemysql_user = 'root'
mysql_password = 'root'
mysql_host = '127.0.0.1'
mysql_port = 3306
mysql_db = 'db'SQLALCHEMY_DATABASE_URL = f'mysql+pymysql://{mysql_user}:{parse.quote_plus(mysql_password)}@{mysql_host}:{mysql_port}/{mysql_db}?charset=utf8'engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_recycle=7200)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)@contextmanager
def session_maker():try:db: Session = SessionLocal()yield dbdb.commit()except Exception as e:print(e)db.rollback()finally:db.close()def get_table_model(table_name, date=None):autobase = automap_base()autobase.prepare(engine, reflect=True)print(autobase)base_model = Nonetry:base_model = getattr(autobase.classes, table_name)except Exception as e:print(e)return base_modelif date == None: # 不分表的情况return base_modelelse: # 分表的情况# 获取基础类base_table = autobase.metadata.tables[table_name]mdata = MetaData()new_table = base_table.tometadata(metadata=mdata, name=f'{table_name}_{date}')# new_table = base_model.tometadata(metadata=mdata, name=f'{table_name}_{date}')try:# 创建表new_table.create(bind=engine)except BaseException as e:# 忽略建表异常，存在多进程同时建表情况，忽略后刷新autobase再试print(e)'''Automap的映射虽然是自动的，但是只有在启动的时候生效，也就是说如果新建一个数据表，而没有告诉Automap，那这个表是找不到的。在实际使用中，可以捕获AttributeError异常，并再次调用AutoBase.prepare(engine, reflect=True) 刷新映射关系。'''autobase = automap_base()autobase.prepare(engine, reflect=True)base_model = getattr(autobase.classes, f'{table_name}_{date}')return base_modeldef get_table_model(table_name, date=None):autobase = automap_base()autobase.prepare(engine, reflect=True)print(autobase)base_model = Nonetry:base_model = getattr(autobase.classes, table_name)except Exception as e:print(e)return base_modelif date == None: # 不分表的情况return base_modelelse: # 分表的情况# 获取基础类base_table = autobase.metadata.tables[table_name]mdata = MetaData()new_table = base_table.tometadata(metadata=mdata, name=f'{table_name}_{date}')# new_table = base_model.tometadata(metadata=mdata, name=f'{table_name}_{date}')try:# 创建表new_table.create(bind=engine)except BaseException as e:# 忽略建表异常，存在多进程同时建表情况，忽略后刷新autobase再试print(e)'''Automap的映射虽然是自动的，但是只有在启动的时候生效，也就是说如果新建一个数据表，而没有告诉Automap，那这个表是找不到的。在实际使用中，可以捕获AttributeError异常，并再次调用AutoBase.prepare(engine, reflect=True) 刷新映射关系。'''autobase = automap_base()autobase.prepare(engine, reflect=True)base_model = getattr(autobase.classes, f'{table_name}_{date}')return base_model# 使用
Record = database.get_table_model("record", '202406')ll = Record(name="1111")with database.session_maker() as db:db.add(ll)db.commit()ll = db.query(Record).all()print(ll)参考：sqlalchemy分表操作 - 知乎
)
