SQLAlchemy使用paginate()返回关联表字段的问题及解决

现有模型定义 #

三个SQLAlchemy模型代码如下：

class User(db.Model):
    __tablename__ = "tbl_users"
    id = db.Column(db.BigInteger, primary_key=True)
    type_id = db.Column(db.Integer, db.ForeignKey("tbl_user_type.id"))
    first_name = db.Column(db.String, nullable=False)
    last_name = db.Column(db.String, nullable=False)
    email = db.Column(db.String, nullable=False, unique=True)
    password = db.Column(db.String, nullable=False)
    user_logs = db.relationship("DashboardLog", backref="log_owner")

class DashboardLog(db.Model):
    __tablename__ = "tbl_logs_dashboard"
    id = db.Column(db.BigInteger, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("tbl_users.id"))
    event_type = db.Column(db.Integer, db.ForeignKey("tbl_dashboard_event_types.id"))
    table_name = db.Column(db.String)
    search_key = db.Column(db.String)
    event_time = db.Column(db.DateTime)

class DashboardEventType(db.Model):
    __tablename__ = "tbl_dashboard_event_types"
    id = db.Column(db.BigInteger, primary_key=True)
    type_name = db.Column(db.String)
    logs = db.relationship("DashboardLog", backref="event_type_id")

原查询及返回结果 #

最初的关联查询代码：

logs = model.query.order_by(time_attr.desc()).join(User).join(DashboardEventType).paginate(page=page, per_page=per_page)

data = []
for log in logs.items:
  log = log.__dict__
  log = {k:v for k,v in log.items() if k != "_sa_instance_state"}
  data.append(log)
return data

返回结果仅包含DashboardLog自身字段：

[
    {
        "event_time": "2022-09-22 13:45:48",
        "event_type": 6,
        "id": 13,
        "search_key": null,
        "table_name": null,
        "user_id": 3
    },
    {
        "event_time": "2022-09-22 13:18:57",
        "event_type": 1,
        "id": 12,
        "search_key": null,
        "table_name": null,
        "user_id": 3
    }
]

需求说明 #

需要将返回结果中的user_id替换为关联User表的email字段，event_type替换为关联DashboardEventType表的type_name字段，对应的原生SQL：

SELECT
tbl_logs_dashboard.table_name AS tbl_logs_dashboard_table_name, 
tbl_logs_dashboard.search_key AS tbl_logs_dashboard_search_key, 
tbl_logs_dashboard.event_time AS tbl_logs_dashboard_event_time,
tbl_users.email,
tbl_dashboard_event_types.type_name
FROM tbl_logs_dashboard JOIN tbl_users ON tbl_users.id = tbl_logs_dashboard.user_id 
JOIN tbl_dashboard_event_types ON tbl_dashboard_event_types.id = tbl_logs_dashboard.event_type 
ORDER BY tbl_logs_dashboard.event_time DESC

错误尝试 #

直接在query()中传入目标字段，触发BaseQuery object is not callable错误：

logs = model.query(User.email, DashboardEventType.type_name).order_by(time_attr.desc()).join(User).join(DashboardEventType).paginate(page=page, per_page=per_page)

正确实现方式 #

使用add_columns()方法添加关联表字段，查询代码如下：

logs = model.query.order_by(time_attr.desc())\
        .join(User)\
        .join(DashboardEventType)\
        .add_columns(User.email, DashboardEventType.type_name)\
        .paginate(page=page, per_page=per_page)

结果处理调整 #

此时logs.items中的每个元素是包含DashboardLog实例、email、type_name的元组，需修改结果处理逻辑：

data = []
for log_item in logs.items:
    log_instance, user_email, event_type_name = log_item
    log_dict = {k:v for k,v in log_instance.__dict__.items() if k != "_sa_instance_state"}
    # 替换字段
    del log_dict["user_id"]
    del log_dict["event_type"]
    log_dict["email"] = user_email
    log_dict["type_name"] = event_type_name
    data.append(log_dict)
return data

内容的提问来源于stack exchange，提问作者m.yagmur