如何用PyMongo实现MongoDB双查询交集获取目标regno

需求说明 #

需要提取同时符合以下两个条件的regno，并获取这些regno对应的所有文档：

• 该regno存在2018年3月31日前且dormant: true的记录
• 该regno存在2020年3月31日后且dormant: false的记录

示例数据中仅0001符合要求。

数据集示例 #

{
  "_id": "uniqueid1",
  "date": "2017-05-30",
  "dormant": true,
  "regno": "0001"
}
{
  "_id": "uniqueid2",
  "date": "2017-05-30",
  "dormant": true,
  "regno": "0002"
}
{
  "_id": "uniqueid3",
  "date": "2017-05-30",
  "dormant": false,
  "regno": "0003"
}
{
  "_id": "uniqueid4",
  "date": "2021-05-30",
  "dormant": false,
  "regno": "0001"
}
{
  "_id": "uniqueid5",
  "date": "2021-05-30",
  "dormant": true,
  "regno": "0002"
}
{
  "_id": "uniqueid6",
  "date": "2021-05-30",
  "dormant": false,
  "regno": "0004"
}

当前代码问题 #

你使用$or逻辑查询，只能拿到满足任一条件的文档，无法筛选出同时符合两个条件的regno的所有记录。

解决方案 #

方法一：分步查询（易上手） #

分四步完成，逻辑清晰，适合新手理解：

1. 找出所有满足第一个条件的regno
2. 找出所有满足第二个条件的regno
3. 取两个集合的交集，得到同时符合要求的regno
4. 查询这些regno对应的所有文档

from pymongo import MongoClient
import datetime

# 连接数据库（替换为你的数据库配置）
client = MongoClient()
db = client.your_database
filed_accounts = db.your_collection

# 定义日期阈值（提前转成datetime类型，避免重复计算）
date_2018 = datetime.datetime(2018, 3, 31)
date_2020 = datetime.datetime(2020, 3, 31)

# 步骤1：获取2018-03-31前 dormant=true 的regno集合
query1 = {
    "$and": [
        {"$expr": {"$lt": [{"$dateFromString": {"dateString": "$date"}}, date_2018]}},
        {"dormant": True}
    ]
}
regnos_cond1 = set(doc["regno"] for doc in filed_accounts.find(query1, {"regno": 1, "_id": 0}))

# 步骤2：获取2020-03-31后 dormant=false 的regno集合
query2 = {
    "$and": [
        {"$expr": {"$gt": [{"$dateFromString": {"dateString": "$date"}}, date_2020]}},
        {"dormant": False}
    ]
}
regnos_cond2 = set(doc["regno"] for doc in filed_accounts.find(query2, {"regno": 1, "_id": 0}))

# 步骤3：取交集得到目标regno
target_regnos = regnos_cond1 & regnos_cond2

# 步骤4：查询目标regno的所有文档
final_docs = list(filed_accounts.find({"regno": {"$in": list(target_regnos)}}))

# 打印结果
for doc in final_docs:
    print(doc)

方法二：聚合框架一步到位（高效） #

用MongoDB聚合管道一次性完成分组判断和文档提取，减少多次查询的网络开销，数据量大时性能更优：

from pymongo import MongoClient
import datetime

client = MongoClient()
db = client.your_database
filed_accounts = db.your_collection

date_2018 = datetime.datetime(2018, 3, 31)
date_2020 = datetime.datetime(2020, 3, 31)

pipeline = [
    # 临时将字符串日期转为datetime类型，方便后续比较
    {
        "$addFields": {
            "date_obj": {"$dateFromString": {"dateString": "$date"}}
        }
    },
    # 按regno分组，判断每个regno是否满足两个条件
    {
        "$group": {
            "_id": "$regno",
            "has_pre_2018_dormant": {
                "$max": {"$and": [{"$lt": ["$date_obj", date_2018]}, {"$eq": ["$dormant", True]}]}
            },
            "has_post_2020_active": {
                "$max": {"$and": [{"$gt": ["$date_obj", date_2020]}, {"$eq": ["$dormant", False]}]}
            },
            # 收集该regno的所有原始文档
            "all_docs": {"$push": "$$ROOT"}
        }
    },
    # 筛选同时满足两个条件的regno分组
    {
        "$match": {
            "has_pre_2018_dormant": True,
            "has_post_2020_active": True
        }
    },
    # 展开文档列表，还原为单个文档格式
    {
        "$unwind": "$all_docs"
    },
    # 将原始文档设为根文档，去掉分组字段
    {
        "$replaceRoot": {"newRoot": "$all_docs"}
    },
    # 移除临时添加的date_obj字段（可选）
    {
        "$project": {"date_obj": 0}
    }
]

final_docs = list(filed_accounts.aggregate(pipeline))

# 打印结果
for doc in final_docs:
    print(doc)

两种方法对比 #

• 分步查询：逻辑清晰，代码简单，适合新手调试，但数据量大时会有多次查询的额外开销。
• 聚合框架：一次查询完成所有操作，性能更优，但需要理解聚合管道的逻辑。

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