Flask+SQLAlchemy按日期范围查询子记录问题求助

针对你遇到的关联查询返回所有TestResult、懒加载不保留时间条件的问题，以下是几种可行的查询方案：

方案一：使用contains_eager主动加载并过滤关联数据 #

通过先关联表再用contains_eager指定过滤条件，确保一次查询就获取符合时间范围的关联数据，彻底避免懒加载的问题：

from datetime import datetime, timedelta
from sqlalchemy.orm import contains_eager
from your_app import db, Endpoint, TestCase, TestResult

# 计算24小时前的时间节点（建议用UTC时间避免时区偏差）
cutoff_time = datetime.utcnow() - timedelta(hours=24)

# 查询包含最近24小时测试结果的Endpoint及关联数据
endpoints = db.session.query(Endpoint)\
    # 依次关联TestCase和TestResult表
    .join(Endpoint.test_cases)\
    .join(TestCase.test_results)\
    # 过滤出最近24小时的测试结果
    .filter(TestResult.created_at >= cutoff_time)\
    # 指定主动加载关联数据，并绑定过滤条件
    .options(
        contains_eager(Endpoint.test_cases)
        .contains_eager(TestCase.test_results)
    )\
    # 去重，避免同一个Endpoint因多条TestResult被重复返回
    .distinct()\
    .all()

查询完成后，每个Endpoint下的test_cases仅包含有符合条件测试结果的用例，每个TestCase下的test_results也只保留最近24小时的记录，直接序列化即可得到目标数据。

方案二：直接查询符合条件的TestResult并关联上级表 #

如果仪表盘核心是展示测试结果数据，可以直接以TestResult为查询主体，关联TestCase和Endpoint，逻辑更直观：

from datetime import datetime, timedelta
from sqlalchemy.orm import joinedload
from your_app import db, TestResult

cutoff_time = datetime.utcnow() - timedelta(hours=24)

# 查询最近24小时的测试结果，同时主动加载关联的TestCase和Endpoint
recent_results = db.session.query(TestResult)\
    .join(TestResult.test_case)\
    .join(TestCase.endpoint)\
    .filter(TestResult.created_at >= cutoff_time)\
    # 主动加载关联对象，避免后续触发懒加载
    .options(
        joinedload(TestResult.test_case).joinedload(TestCase.endpoint)
    )\
    .all()

后续可以按Endpoint和TestCase对recent_results进行分组，再序列化展示到仪表盘。

方案三：子查询预过滤TestResult #

先通过子查询获取最近24小时的TestResult ID集合，再关联查询上级表，适合复杂关联场景：

from datetime import datetime, timedelta
from sqlalchemy.orm import contains_eager
from your_app import db, Endpoint, TestCase, TestResult

cutoff_time = datetime.utcnow() - timedelta(hours=24)

# 子查询：获取最近24小时的TestResult ID
recent_result_ids = db.session.query(TestResult.id)\
    .filter(TestResult.created_at >= cutoff_time)\
    .subquery()

# 关联查询Endpoint，仅包含有符合条件TestResult的记录
endpoints = db.session.query(Endpoint)\
    .join(Endpoint.test_cases)\
    .join(TestCase.test_results)\
    .filter(TestResult.id.in_(recent_result_ids))\
    .options(
        contains_eager(Endpoint.test_cases)
        .contains_eager(TestCase.test_results)
    )\
    .distinct()\
    .all()

关键注意点 #

• 禁用懒加载：懒加载会在访问关联属性时单独发起查询，不会继承主查询的过滤条件，必须使用主动加载（joinedload/contains_eager）。
• 时区统一：建议用UTC时间计算cutoff_time，避免服务器时区与数据库时区不一致导致的过滤偏差。

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