如何用Peewee基于其他表计算结果创建并填充新表？

我明白你遇到的问题了——用Peewee做复杂聚合后生成新表确实不像Pandas那么直观，但完全可以实现！下面我一步步给你演示怎么完成这个需求：

1. 先定义现有表的Peewee模型 #

首先确保你已经正确映射了User和Sessions表的模型（如果还没定义的话）：

from peewee import *

# 连接你的数据库
db = SqliteDatabase('database.db')

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

class User(BaseModel):
    id = IntegerField(primary_key=True)
    name = CharField()

class Sessions(BaseModel):
    id = IntegerField()  # 关联User表的id
    session_id = CharField(column_name='SessionId')  # 映射原表的SessionId字段

2. 编写聚合查询获取统计数据 #

要得到sessionCount（总会话数）和TopSession（出现最多的会话ID），我们需要用到子查询和窗口函数：

步骤2.1：统计每个用户每个会话的出现次数 #

先分组统计每个用户的每个SessionId出现了多少次：

# 子查询1：统计用户-会话的出现次数
session_freq = (Sessions
                .select(Sessions.id, Sessions.session_id,
                        fn.COUNT(Sessions.session_id).alias('freq'))
                .group_by(Sessions.id, Sessions.session_id))

步骤2.2：给每个用户的会话按出现次数排名 #

用窗口函数ROW_NUMBER()给每个用户的会话按频次降序排名，这样排名第一的就是出现最多的会话：

# 子查询2：给每个用户的会话频次排名，取Top1
ranked_sessions = (session_freq
                   .select(session_freq.id, session_freq.session_id,
                           fn.ROW_NUMBER().over(
                               partition_by=session_freq.id,
                               order_by=[fn.DESC(session_freq.c.freq)]
                           ).alias('rank')))

步骤2.3：关联用户表获取完整统计数据 #

最后关联User表，拿到用户的基本信息、总会话数和Top会话：

# 主查询：整合用户信息、总会话数、Top会话
user_stats = (User
              .select(User.id, User.name,
                      fn.COUNT(Sessions.id).alias('sessionCount'),
                      ranked_sessions.c.session_id.alias('TopSession'))
              .join(Sessions, on=(User.id == Sessions.id))
              .join(ranked_sessions, on=(User.id == ranked_sessions.c.id) & (ranked_sessions.c.rank == 1))
              .group_by(User.id, User.name, ranked_sessions.c.session_id))

3. 创建新表并插入统计数据 #

现在定义新表UserSessionStats的模型，然后把查询结果插入进去：

# 定义新表模型
class UserSessionStats(BaseModel):
    id = IntegerField(primary_key=True)
    name = CharField()
    sessionCount = IntegerField()
    TopSession = CharField()

# 创建表（safe=True表示如果表已存在则跳过）
UserSessionStats.create_table(safe=True)

# 批量插入统计数据（用atomic保证事务安全）
with db.atomic():
    # 更高效的批量插入方式
    stats_data = [
        {
            'id': s.id,
            'name': s.name,
            'sessionCount': s.sessionCount,
            'TopSession': s.TopSession
        } for s in user_stats
    ]
    UserSessionStats.insert_many(stats_data).execute()

关键补充说明 #

• 如果你的Sessions表中，多个SessionId出现次数并列最多，上面的方法会随机取其中一个。如果需要保留所有并列的Top会话，可以把ROW_NUMBER()换成RANK()，然后调整查询逻辑筛选排名为1的所有结果。
• 如果你使用的是MySQL等不支持窗口函数的旧版本数据库，可以改用子查询嵌套的方式实现TopSession的统计，逻辑核心是找出每个用户会话频次的最大值，再关联对应的SessionId。

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