如何从字符串获取SQLAlchemy表及字段对象并执行ORM删除操作

针对你提出的两个问题，直接给你可行的实现方案：

1. 通过字符串获取表对象 #

SQLAlchemy分Core和ORM两种使用模式，获取表对象的方式略有不同：

• ORM模式（用继承declarative_base的模型类）：
  可以自己维护一个「表名-模型类」的映射字典，或者直接用Base的classes属性获取（前提是所有模型都基于同一个Base创建）。示例：

  # 假设已定义TableA、TableB模型，均继承自Base
  model_map = {"tableA": TableA, "tableB": TableB}
  model_cls = model_map[table_name]
  # 或者用Base内置的映射
  model_cls = Base.classes.get(table_name)
  

• Core模式（直接用Table类定义表结构）：
  从MetaData对象的tables字典里按表名取即可：

  table_obj = metadata.tables[table_name]
  

2. 通过字段名字符串执行ORM删除查询 #

拿到模型/表对象后，用getattr()函数就能通过字段名获取对应的字段对象，之后构造删除条件执行批量删除即可。注意SQLAlchemy的批量删除和单对象删除是不同的API：

• ORM批量删除：用session.query(模型类).filter(条件).delete()，返回值就是删除的记录数
• Core批量删除：构造delete()语句，执行后通过结果的rowcount属性获取删除数量

完整修正后的示例代码 #

假设你用的是ORM模式，基于declarative_base定义模型：

from sqlalchemy import create_engine, Column, DateTime, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timedelta

# 初始化ORM基础
Base = declarative_base()

# 定义模型
class TableA(Base):
    __tablename__ = "tableA"
    id = Column(Integer, primary_key=True)
    modified_on = Column(DateTime)

class TableB(Base):
    __tablename__ = "tableB"
    id = Column(Integer, primary_key=True)
    event_on = Column(DateTime)

# 创建数据库会话
engine = create_engine("sqlite:///your_db.db")
Session = sessionmaker(bind=engine)
session = Session()

# 修复后的配置（补全原代码缺失的逗号）
json_config = [
    {"table": "tableA", "field": "modified_on", "expires": 30},
    {"table": "tableB", "field": "event_on", "expires": 30}
]

# 表名到模型的映射
model_map = {"tableA": TableA, "tableB": TableB}

for item in json_config:
    table_name = item["table"]
    field_name = item["field"]
    expire_days = item["expires"]
    
    # 获取模型类
    model_cls = model_map.get(table_name)
    if not model_cls:
        print(f"跳过不存在的表：{table_name}")
        continue
    
    # 获取字段对象
    field = getattr(model_cls, field_name, None)
    if not field:
        print(f"表 {table_name} 不存在字段 {field_name}，跳过")
        continue
    
    # 计算过期时间（当前时间往前推expire_days天）
    cutoff_time = datetime.now() - timedelta(days=expire_days)
    
    # 执行批量删除，synchronize_session=False提升性能
    deleted_count = session.query(model_cls).filter(field < cutoff_time).delete(synchronize_session=False)
    
    session.commit()
    print(f"{table_name}: 移除了 {deleted_count} 条记录")

原示例代码的问题说明 #

1. JSON配置中每个键值对末尾缺少逗号，会触发语法错误
2. session.delete()是用来删除单个ORM对象的，批量删除需要用query.filter().delete()
3. ORM删除操作直接返回删除的记录数，不需要从result.row_count获取（Core模式下用result.rowcount）

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