如何使用Python SQLAlchemy声明式基类正确获取动态表记录

核心问题修正 #

原写法存在两个致命问题，会直接导致功能不可用：

• 每次调用函数时在内部重复定义模型类，会重复向SQLAlchemy声明式基类的元数据注册映射，第二次执行就会触发类重复注册的报错
• 原逻辑返回的是模型类本身，而非查询到的记录实例，无法直接完成字段读取、修改操作

完整实现代码 #

首先前置依赖初始化，替换成你实际的数据库连接配置即可：

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy import Column, Integer, String

# 初始化异步数据库连接，替换为你的实际数据库DSN
engine = create_async_engine("mysql+aiomysql://user:pass@host/dbname")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# 通用CRUD混入，方便后续修改记录后持久化
class CRUDMixin:
    async def save(self):
        async with async_session() as session:
            session.add(self)
            await session.commit()
            await session.refresh(self)

# 动态模型缓存，避免重复映射冲突
_model_cache = {}
def get_guild_user_model(guild_id: int):
    if guild_id in _model_cache:
        return _model_cache[guild_id]
    
    # 动态生成对应分表的模型类
    class GuildUser(Base, CRUDMixin):
        __tablename__ = f"guild_{guild_id}"
        id = Column(Integer, primary_key=True) # 对应传入的user_id
        description = Column(String)
        sex = Column(Integer)
        age = Column(Integer)
    
    _model_cache[guild_id] = GuildUser
    return GuildUser

# 实现目标查询函数
async def get_user(guild_id: int, user_id: int):
    UserModel = get_guild_user_model(guild_id)
    async with async_session() as session:
        # 按主键查询对应用户记录，如果user_id不是主键请改用select条件查询
        user = await session.get(UserModel, user_id)
        return user

调用示例 #

完全匹配你预期的使用方式：

async def main():
    user = await get_user(12345, 123)
    # 读取字段值
    print(user.description)
    # 修改字段值
    user.description = "Hello"
    # 提交修改到数据库
    await user.save()
    print(user.description)

asyncio.run(main())

注意事项 #

• 所有guild_%number%格式的分表结构必须和模型定义的字段、类型完全一致，否则查询时会抛出字段匹配错误
• 如果存在动态新建分表的需求，可以在get_guild_user_model生成类之后，添加建表逻辑：

  async with engine.begin() as conn:
      await conn.run_sync(Base.metadata.create_all, tables=[GuildUser.__table__])
  

• 如果你的user_id不是对应表的id主键，把session.get的查询逻辑改成select(UserModel).where(UserModel.对应字段 == user_id)即可
• 模型缓存是必须的，禁止跳过缓存每次重复定义模型类，否则会触发SQLAlchemy的映射冲突异常

内容的提问来源于stack exchange，提问作者Mr. Nom4ik