SQLAlchemy按即将到来的生日/周年纪念筛选

Question

环境：python 3.8，炼金术，postgres

class User(db.Model):
    name = db.Column(db.Text)
    birthday = db.Column(db.DateTime)

    @classmethod
    def upcoming_birthdays(cls):
        return (cls.query
                .filter("??")
                .all()
                )

我想创建一个sqlalchemy查询，用于过滤X天内即将到来的生日的用户。我曾考虑过使用extract函数从生日中提取月份和日期，但在月底或年底的几天内都不起作用。我也想过试着把生日改成朱利安式的日期来做比较，但我不知道该怎么做

例如，如果今天是2020年8月30日，它将返回用户的生日

• 1995年9月1日
• 2010年8月31日。。。。等

谢谢你的帮助

Answer 1

您可以实现以下简单查询的目标：

q = (
    db.session.query(User)
    .filter(has_birthday_next_days(User.birthday, 7))
)

这不是User上的@classmethod，但是如果您愿意，可以将解决方案转换为@classmethod

---

剩下要做的就是实际实现has_birthday_next_days(...)，下面列出了该原则，主要是该原则的文档：

def has_birthday_next_days(sa_col, next_days: int = 0):
    """ 
    sqlalchemy expression to indicate that an sa_col (such as`User.birthday`)
    has anniversary within next `next_days` days.

    It is implemented by simply checking if the 'age' of the person (in years)
    has changed between today and the `next_days` date.
    """
    return age_years_at(sa_col, next_days) > age_years_at(sa_col)

可以有多个age_years_at的实现，下面只是一种可能性，特别是postgresql（包括所需的导入）：

import datetime
import sqlalchemy as sa

def age_years_at(sa_col, next_days: int = 0):
    """
    Generates a postgresql specific statement to return 'age' (in years)'
    from an provided field either today (next_days == 0) or with the `next_days` offset.
    """
    stmt = func.age(
        (sa_col - sa.func.cast(datetime.timedelta(next_days), sa.Interval))
        if next_days != 0
        else sa_col
    )
    stmt = func.date_part("year", stmt)
    return stmt

最后，所需的查询q = db.session.query(User).filter(has_birthday_next_days(User.birthday, 30))生成：

SELECT "user".id,
       "user".name,
       "user".birthday
FROM   "user"
WHERE   date_part(%(date_part_1)s, age("user".birthday - CAST(%(param_1)s AS INTERVAL))) 
      > date_part(%(date_part_2)s, age("user".birthday))

{'date_part_1': 'year', 'param_1': datetime.timedelta(days=30), 'date_part_2': 'year'}

---

好处：通过使用泛型函数实现了这一点，它不仅可以用于User.birthday列，还可以用于任何其他类型兼容的值。此外，这些函数还可以分别用于语句的select和where部分。例如：

q = (
    db.session.query(
        User,
        age_years_at(User.birthday).label("age_today"),
        age_years_at(User.birthday, 7).label("age_in_a_week"),
        has_birthday_next_days(User.birthday, 7).label("has_bday_7-days"),
        has_birthday_next_days(User.birthday, 30).label("has_bday_30-days"),
    )
    .filter(has_birthday_next_days(User.birthday, 30))
)

Answer 2

您最初关于提取日和月的想法很好。
如果您从1995年9月1日生日开始，您可以将该日期和月份移到当前年份（即2020年9月1日），然后检查该日期是否在您指定的范围内（从今天到今天+X天）

至于年底的日子，你可以通过将生日和月份不仅移到今年，而且移到下一年来解决这个问题。
例如，假设今天是2020年12月30日，你的生日是1995年1月2日。然后，您将检查2020年1月2日或2021年1月2日是否在您指定的范围内

from datetime import date, timedelta
from sqlalchemy import func,or_

@classmethod
def upcoming_birthdays(cls):
    dateFrom = date.today()
    dateTo = date.today() + timedelta(days=5)
    thisYear = dateFrom.year
    nextYear = dateFrom.year + 1
    return (cls.query
            .filter(
                or_(
                    func.to_date(func.concat(func.to_char(cls.birthday, "DDMM"), thisYear), "DDMMYYYY").between(dateFrom,dateTo),
                    func.to_date(func.concat(func.to_char(cls.birthday, "DDMM"), nextYear), "DDMMYYYY").between(dateFrom,dateTo)
                )
            )
            .all()
            )