使用SQLAlchemy排查MySQL中员工活动时段重叠问题

核心问题 #

你的SQLAlchemy代码中，时间范围过滤的表达式写法错误，导致无法生成正确的SQL时间计算逻辑：

models.Session.date + (models.Session.duration * timedelta(hours=1)) >= new_session_start

这段代码试图在Python层面用timedelta处理数据库字段，但SQLAlchemy无法将这个操作转换成MySQL的INTERVAL语法，最终生成的SQL中时间条件完全不生效，只有员工ID筛选起作用。

修正方案 #

需要使用SQLAlchemy的func工具，调用MySQL原生的时间计算函数来构建正确的过滤条件，对应原生SQL的session.date + INTERVAL session.duration SECOND，以下是两种可行写法：

写法1：使用func.timestampadd（推荐） #

from sqlalchemy import func

def get_overlapping_staffSession(db: Session, session: schemas.SessionCreate) -> list[models.StaffSession]:
    staff_ids = [staff.id for staff in session.staff]

    new_session_start = session.date
    # 注意：这里要确认Session.duration的单位，原生SQL用的是秒，若你的duration是小时则保留hours，否则改成seconds
    new_session_end = new_session_start + timedelta(hours=session.duration)

    overlapping_sessions = (
        db.query(models.StaffSession)
        .join(models.Session, models.StaffSession.id_session == models.Session.id)
        .filter(
            models.StaffSession.id_staff.in_(staff_ids),
            models.Session.date <= new_session_end,
            # 调用MySQL的timestampadd函数计算现有session的结束时间
            func.timestampadd(func.second, models.Session.duration, models.Session.date) >= new_session_start
        )
    ).all()

    return overlapping_sessions

写法2：直接构造INTERVAL表达式 #

from sqlalchemy import text

def get_overlapping_staffSession(db: Session, session: schemas.SessionCreate) -> list[models.StaffSession]:
    staff_ids = [staff.id for staff in session.staff]

    new_session_start = session.date
    new_session_end = new_session_start + timedelta(hours=session.duration)

    overlapping_sessions = (
        db.query(models.StaffSession)
        .join(models.Session, models.StaffSession.id_session == models.Session.id)
        .filter(
            models.StaffSession.id_staff.in_(staff_ids),
            models.Session.date <= new_session_end,
            # 直接拼接MySQL的INTERVAL语法
            models.Session.date + text("INTERVAL :duration SECOND").bindparams(duration=models.Session.duration) >= new_session_start
        )
    ).all()

    return overlapping_sessions

额外注意点 #

• 确认Session.duration的单位：原生SQL中用的是SECOND，如果你的duration字段存储的是小时数，需要把SQL中的SECOND改成HOUR，同时Python里的timedelta单位也要对应一致，否则时间范围计算会出现偏差。
• 可以用列表推导式简化staff_ids的生成，替代原有的for循环写法。

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