def get_streaks(s: sqlalchemy.orm.session.Session,
active: Optional[bool]=None,
limit: Optional[int]=None,
max_age: Optional[int]=None,
) \
-> Sequence[Streak]:
"""Get streaks, ordered by length (longest first).
Parameters:
active: only return streaks with this active flag
limit: only return (up to) limit results
max_age: only return streaks with a win less than this many days old
Returns:
List of active streaks.
"""
# The following code is a translation of this basic SQL:
# SELECT streaks.*, count(games.streak_id) as streak_length
# FROM streaks
# JOIN games ON (streaks.id = games.streak_id)
# GROUP BY streaks.id
# HAVING streak_length > 1
# ORDER BY streak_length DESC
streak_length = func.count(Game.streak_id).label('streak_length')
streak_last_activity = func.max(Game.end).label('streak_last_activity')
q = s.query(Streak, streak_length).join(Streak.games)
q = q.group_by(Streak.id)
q = q.having(streak_length > 1)
if max_age is not None:
q = q.having(
streak_last_activity > func.date('now', '-%s day' % max_age))
q = q.order_by(streak_length.desc())
if active is not None:
q = q.filter(Streak.active == (sqlalchemy.true()
if active else sqlalchemy.false()))
if limit is not None:
q = q.limit(limit)
streaks = q.all()
# Since we added a column to the query, the result format is:
# ((Streak, length), (Streak, length), ...)
# It's annoying to deal with a custom format, and recalculating the streak
# length for a few streaks is NBD, so just return a list of Streaks
return [t.Streak for t in streaks]
评论列表
文章目录
