如何让SQLAlchemy模型读写未显式映射的数据库列？

现有预存数据库的roles表用于存储用户角色，除id、name等常规列外，还有大量priv_docs、priv_payments这类命名格式的动态权限列。需要创建SQLAlchemy ORM模型与之交互，用于操作现有应用数据库的命令行工具。

现有基础模型如下：

class Role(Base):
    __tablename__ = 'roles'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    notes: Mapped[Optional[str]]

需求限制：

• 工具需适配主应用的不同变体，各变体的priv_*列集合不同，不想在模型中显式映射所有priv_*列
• 希望通过类似role.priv['docs']的字典式接口访问权限列
• 已知__table_args__ = {'autoload_with': engine}的反射方案，但工具需在同一次执行中通过schema_translate_map连接多个数据库，单库反射无法适配其他库

理想工作流程：

1. 执行sess.execute(select(Role))查询
2. SQLAlchemy转换为SELECT * FROM roles
3. 映射模型实例时，所有priv_*列可被实例访问
4. 用@property实现priv字典接口
5. 更新priv_*列后保存实例，数据库对应列同步更新

请问是否存在方法让SQLAlchemy模型读写未显式映射的列？

可以通过SQLAlchemy的实例内置__dict__结合自定义@property实现需求，核心思路是让ORM自动加载所有查询返回的列（包括未显式映射的），并将这些列存入实例的__dict__中，再通过封装属性实现字典式的权限列访问。

实现步骤 #

1. 模型基础配置：
   无需额外复杂映射配置，只需确保执行查询时返回所有列（比如用SELECT *），SQLAlchemy会自动将未显式映射的列存入实例的__dict__。同时开启eager_defaults=True确保更新时能同步未映射列的默认值。

2. 封装priv字典接口：
   通过@property和@priv.setter，自动处理priv_前缀的添加与移除，将实例__dict__中的权限列转换为简洁的字典格式。

完整代码示例 #

from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column, Base, Session
from typing import Optional, Dict, Any

class Role(Base):
    __tablename__ = 'roles'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    notes: Mapped[Optional[str]]
    
    # 确保更新时同步未映射列的默认值
    __mapper_args__ = {
        "eager_defaults": True,
    }

    @property
    def priv(self) -> Dict[str, Any]:
        # 提取所有priv_前缀列，移除前缀作为字典键
        return {
            key[5:]: value 
            for key, value in self.__dict__.items() 
            if key.startswith('priv_') and not key.startswith('_')
        }

    @priv.setter
    def priv(self, value: Dict[str, Any]):
        # 为字典键添加priv_前缀，存入实例__dict__
        for key, val in value.items():
            self.__dict__[f'priv_{key}'] = val

# 使用示例
def example_usage(session: Session):
    # 查询所有角色，自动加载包括priv_*在内的所有列
    roles = session.execute(select(Role)).scalars().all()
    
    for role in roles:
        # 读取权限
        print(f"Role {role.name} docs permission: {role.priv.get('docs')}")
        
        # 更新权限
        role.priv['payments'] = True
        role.priv['docs'] = False
    
    # 提交后，数据库中对应的priv_payments、priv_docs列会同步更新
    session.commit()

额外说明 #

• 执行SELECT *时，SQLAlchemy会自动将所有返回列存入实例__dict__，无需提前映射
• 若需指定查询列而非全量查询，只需确保包含目标priv_*列，ORM同样会将其存入__dict__
• 该方案适配多数据库场景，无需提前反射表结构，只要各数据库的roles表遵循priv_*命名规则即可

内容的提问来源于stack exchange，提问作者DMJ