PyMongo如何按客户获取最近到期保单并按日期排序

核心思路是优先在MongoDB端用聚合管道完成计算，避免拉取全量数据到Python端，可根据Mongo版本选择不同的实现写法。 #

前置注意事项 #

• 你示例里的expiry_date是DD-MM-YYYY格式的字符串，这类格式直接做字符串排序会出现日期计算混乱的问题，生产环境建议把到期日存为Mongo原生Date类型，或者统一存为YYYY-MM-DD格式的字符串，才能保证最值计算、排序结果的准确性。
• 你列的示例期望结果存在顺序偏差：按给出的日期值换算为真实日期后，可通过聚合排序阶段的方向参数直接调整成需要的排序规则，不需要改动核心查询逻辑。

方案1：适配MongoDB 5.2+版本（效率最优） #

新版本支持$top聚合阶段，可以直接在分组时取出组内符合排序规则的完整文档，步骤最少性能最高：

from pymongo import DESCENDING, ASCENDING

# 前置假设：db是已经初始化完成的MongoDB数据库连接对象
aggregate_pipeline = [
    # 按用户ID分组，取每个用户下到期日最晚的完整保单
    {
        "$group": {
            "_id": "$person_id",
            "latest_policy": {
                "$top": {
                    "sortBy": {"expiry_date": DESCENDING}, # 组内按到期日倒序，取第一条
                    "output": "$$ROOT" # 保留整条保单的所有字段
                }
            }
        }
    },
    # 把嵌套的保单文档拆平为顶级字段
    {"$replaceRoot": {"newRoot": "$latest_policy"}},
    # 全局按到期日排序，ASCENDING为正序（早->晚），DESCENDING为倒序（晚->早）
    {"$sort": {"expiry_date": ASCENDING}}
]

# 执行查询获取结果
query_result = list(db.policy.aggregate(aggregate_pipeline))

方案2：适配MongoDB 5.2以下低版本 #

低版本不支持$top阶段，可通过分组取最大到期日，再回查原集合关联完整保单的方式实现：

legacy_pipeline = [
    # 第一步：按用户ID分组，先拿到每个用户的最大到期日
    {
        "$group": {
            "_id": "$person_id",
            "max_expiry": {"$max": "$expiry_date"}
        }
    },
    # 第二步：关联原policy集合，匹配对应用户+最大到期日的完整保单
    {
        "$lookup": {
            "from": "policy",
            "let": {"pid": "$_id", "max_exp": "$max_expiry"},
            "pipeline": [
                {
                    "$match": {
                        "$expr": {
                            "$and": [
                                {"$eq": ["$person_id", "$$pid"]},
                                {"$eq": ["$expiry_date", "$$max_exp"]}
                            ]
                        }
                    }
                }
            ],
            "as": "matched_policy"
        }
    },
    # 把关联出来的数组拆为单条文档
    {"$unwind": "$matched_policy"},
    # 把文档结构替换为完整的保单字段
    {"$replaceRoot": {"newRoot": "$matched_policy"}},
    # 全局按到期日排序
    {"$sort": {"expiry_date": ASCENDING}}
]

legacy_result = list(db.policy.aggregate(legacy_pipeline))

可选扩展：关联person集合获取用户姓名 #

如果需要把person集合的用户姓名和保单信息关联返回，可以在$sort阶段前追加如下聚合阶段：

# 关联person集合
{
    "$lookup": {
        "from": "person",
        "localField": "person_id",
        "foreignField": "_id",
        "as": "user_detail"
    }
},
# 一对一关联，直接拆为单条字段
{"$unwind": "$user_detail"}

小数据量补充方案 #

如果policy集合数据量很小，不想写聚合管道，也可以把数据拉到Python端处理，实现方式更直观，但数据量大时不推荐这种方式（会产生大量数据传输开销）：

# 拉取所有保单
all_policies = list(db.policy.find())
# 按person_id分组存储最晚到期的保单
latest_policy_map = {}
for policy in all_policies:
    pid = policy["person_id"]
    # 没存过这个用户，或者当前保单到期日比已存的晚，就更新存储
    if pid not in latest_policy_map or policy["expiry_date"] > latest_policy_map[pid]["expiry_date"]:
        latest_policy_map[pid] = policy
# 把结果转成列表并按到期日排序
final_result = sorted(latest_policy_map.values(), key=lambda x: x["expiry_date"])

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