如何用Peewee实现多关联表数据按时间戳统一排序查询？

需求说明 #

需要从User关联的Tweet和Facebook表中取出某用户的所有活动（推文、帖子），并按时间戳降序（最新优先）混合展示，且无法将数据合并到单个Activity表。

现有问题 #

当前代码只能分别对推文和帖子排序展示，无法将两类记录混合后统一按时间排序。

解决方案 #

利用Peewee的Union操作，将两个表的查询结果合并为一个结构统一的数据集，再按时间戳排序。核心是让两个查询返回相同字段（可额外添加标识字段区分活动来源），再执行联合查询并排序。

修改后的完整代码 #

import datetime
from peewee import *

db = SqliteDatabase(':memory:')

class BaseModel(Model):
    class Meta:
        database = db

class User(BaseModel):
    username = TextField()

    @property
    def tweets_date_desc(self):
        return self.tweets.order_by(Tweet.timestamp.desc())

    @property
    def posts_date_desc(self):
        return self.posts.order_by(Facebook.timestamp.desc())

    @property
    def all_activities_date_desc(self):
        # 构造推文查询：指定返回字段，添加来源标识
        tweet_query = (Tweet
                       .select(Tweet.timestamp, Tweet.content, 
                               Value('tweet').alias('source'))
                       .where(Tweet.user == self))
        # 构造脸书帖子查询：保持与推文查询相同的返回字段结构
        fb_query = (Facebook
                    .select(Facebook.timestamp, Facebook.content, 
                            Value('facebook').alias('source'))
                    .where(Facebook.user == self))
        # 合并两个查询，按时间戳降序排序
        union_query = tweet_query.union(fb_query).order_by(SQL('timestamp').desc())
        return union_query

class Tweet(BaseModel):
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)
    user = ForeignKeyField(User, backref='tweets')

class Facebook(BaseModel):
    content = TextField()
    timestamp = DateTimeField(default=datetime.datetime.now)
    user = ForeignKeyField(User, backref='posts')

测试与使用示例 #

# 初始化数据库并插入测试数据
db.connect()
db.create_tables([User, Tweet, Facebook])

# 创建测试用户
user = User.create(username='huey')

# 插入不同时间的活动记录
Tweet.create(user=user, content='第一条推文', timestamp=datetime.datetime(2024, 5, 1, 10, 0))
Facebook.create(user=user, content='第一条脸书帖子', timestamp=datetime.datetime(2024, 5, 1, 11, 0))
Tweet.create(user=user, content='第二条推文', timestamp=datetime.datetime(2024, 5, 1, 12, 0))
Facebook.create(user=user, content='第二条脸书帖子', timestamp=datetime.datetime(2024, 5, 1, 9, 0))

# 获取并展示混合排序的活动
print(f"用户 {user.username} 的所有活动（按时间降序）：")
for activity in user.all_activities_date_desc:
    print(f"[{activity.source}] {activity.timestamp} - {activity.content}")

关键要点 #

• 两个子查询必须返回相同数量和类型的字段，这里统一选择timestamp、content，并添加source字段区分活动类型（可选，但便于识别来源）
• 使用union()合并查询结果，默认会自动去重；如果需要保留重复记录，可改用union_all()
• 合并后的数据集直接通过order_by()按时间戳降序排序，实现统一展示

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