FastApi结合Motor操作MongoDB 按字段计算结果筛选聚合数据

推荐方案：使用$expr运算符实现同文档字段运算比较 #

MongoDB 原生提供$expr查询运算符，支持在查询条件中嵌入聚合表达式，直接对同一文档的多个字段执行运算后完成条件匹配，不需要额外拉取数据到应用侧计算，性能最优。

代码实现示例 #

import time
from motor.motor_asyncio import AsyncIOMotorCollection

async def fetch_pending_tasks(task_collection: AsyncIOMotorCollection):
    # 生成当前秒级时间戳，与last_run单位保持一致
    current_timestamp = int(time.time())
    filter_query = {
        "$expr": {
            # 比较逻辑：last_run + interval < 当前时间戳
            "$lt": [
                {"$add": ["$last_run", "$interval"]},
                current_timestamp
            ]
        }
    }
    # 异步获取所有符合条件的文档
    cursor = task_collection.find(filter_query)
    return await cursor.to_list(length=None)

性能优化建议 #

如果集合数据量较大，可以增加前置过滤条件并搭配索引大幅提升查询效率：

1. 先给last_run字段创建单字段索引
2. 调整查询条件，先过滤掉last_run本身已经大于当前时间戳的文档（这部分文档的last_run + interval必然大于当前时间，不需要参与后续计算）

filter_query = {
    # 前置过滤命中索引，减少后续$expr的计算量
    "last_run": {"$lt": current_timestamp},
    "$expr": {
        "$lt": [{"$add": ["$last_run", "$interval"]}, current_timestamp]
    }
}

单位适配说明 #

如果你的last_run存储的是毫秒级时间戳，需要调整运算逻辑对齐单位：

filter_query = {
    "$expr": {
        "$lt": [
            # interval单位为秒，转毫秒再和last_run相加
            {"$add": ["$last_run", {"$multiply": ["$interval", 1000]}]},
            int(time.time() * 1000)
        ]
    }
}

备选方案：预计算下次运行时间（超高吞吐场景适用） #

如果你的定时任务查询频率极高、集合数据量超过百万级，可以采用空间换时间的方案：

• 新增next_run字段，每次写入/更新last_run或interval时，同步预计算next_run = last_run + interval的值存入该字段
• 查询时直接使用普通匹配条件，可完美命中next_run索引，查询性能最高

# 查询逻辑极简，性能最优
filter_query = {"next_run": {"$lt": current_timestamp}}

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