Python 使用mySQL的SQlalchemy过滤重叠时间段

根据INTERVAL类型的SQLAchemy docs:

datetime. timedela()对象的类型.

Interval类型处理datetime. timeta对象.在PostgreSQL和Oracle中，使用原生INTERVAL类型;对于其他类型，该值存储为相对于"纪元"(1970年1月1日)的日期.

请注意，Interval类型目前在不本地支持Interval类型的平台上不提供日期算术运算.此类操作通常需要转换运算式的两侧(例如，首先将两侧转换为整纪元值)，这目前是一个手动过程(例如通过express. full).

因此，使用间隔的日期时间算术在SQL上不直接支持.我们可以通过在一些示例日期上执行问题中的查询的间隔算术来验证这一点:

import sqlalchemy as sa

engine = sa.create_engine(...)
sessions = sa.Table(...)

# (create the table, insert some dates...)

QUERY = """
SELECT 
    date + INTERVAL duration SECOND
FROM 
    sessions
ORDER BY id
"""

with engine.connect() as conn:
    rows = conn.execute(sa.text(QUERY))
    for row in rows:
        print(row)

输出

(datetime.datetime(2024, 4, 15, 17, 30),)
(datetime.datetime(2024, 4, 15, 17, 0),)
(datetime.datetime(2024, 4, 15, 13, 0),)

但SQLAchemy查询

with engine.connect() as conn:
    q = sa.select(
        sessions.c.date + (sessions.c.duration * dt.timedelta(hours=1))
    ).order_by(sessions.c.id)
    rows = conn.execute(q)
    for row in rows:
        print(row)

输出

(20240422255000.0,)
(20240422252000.0,)
(20240422212000.0,)

至少有两种方法可以解决这个问题.我们可以将所有元素转换为UNIX时间戳并返回，或者我们可以使用sqlalchemy.text来插入所需的SQL.

with engine.connect() as conn:
    q = sa.select(
        sa.func.from_unixtime(
            sa.func.unix_timestamp(sessions.c.date) + (sessions.c.duration)
        ),
        sessions.c.date + sa.text('INTERVAL sessions.duration SECOND')
        
    ).order_by(sessions.c.id)
    rows = conn.execute(q)
    for row in rows:
        print(row)

给我们

(datetime.datetime(2024, 4, 15, 17, 30), datetime.datetime(2024, 4, 15, 17, 30))
(datetime.datetime(2024, 4, 15, 17, 0), datetime.datetime(2024, 4, 15, 17, 0))
(datetime.datetime(2024, 4, 15, 13, 0), datetime.datetime(2024, 4, 15, 13, 0))

上面的示例使用SQLAlchemy的核心语法. 等效的ORM语法是

overlapping_sessions = (
    db.query(models.StaffSession)
    .join(models.Session, models.StaffSession.id_session == models.Session.id)
    .filter(
        and_(
            models.StaffSession.id_staff.in_(staff_ids),
            models.Session.date <= new_session_end,
            sa.func.from_unixtime(sa.func.unix_timestamp(models.Sessions.date) + (model.Session.duration)) >= new_session_start
        )
    )
).all()