基于现有MongoDB连接代码，如何编写数据验证与值查询方法？

先优化原有连接逻辑，改成可复用的工具函数，方便后续验证方法调用：

from pymongo import MongoClient
import urllib.parse
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

def get_mongo_collection():
    user = os.environ['USER']
    password = urllib.parse.quote_plus(os.environ['PASS'])
    # 修正连接字符串的语法错误
    mongo_client = MongoClient(f'mongodb://{user}:{password}@hostname:portname/Database?tlsAllowInvalidCertificates=True&connect=False')
    db = mongo_client.get_database("Database")
    return db.get_collection("Collection")

以下是针对{"AcctEntxxxx"}这类查询的实用验证方法：

1. 验证指定条件的文档是否存在 #

快速判断符合条件的文档是否在集合中：

def check_document_exists(query):
    collection = get_mongo_collection()
    exists = collection.find_one(query) is not None
    print(f"符合条件的文档{'存在' if exists else '不存在'}")
    return exists

# 调用示例：检查包含AcctEntxxxx字段的文档是否存在
check_document_exists({"AcctEntxxxx": {"$exists": True}})
# 检查AcctEntxxxx等于特定值的文档是否存在
check_document_exists({"AcctEntxxxx": "目标账号值"})

2. 统计匹配条件的文档总数 #

统计符合查询条件的文档数量：

def count_matching_documents(query):
    collection = get_mongo_collection()
    count = collection.count_documents(query)
    print(f"符合条件的文档总数：{count}")
    return count

# 调用示例：统计所有包含AcctEntxxxx字段的文档数量
count_matching_documents({"AcctEntxxxx": {"$exists": True}})

3. 验证匹配文档的字段完整性 #

检查符合条件的文档是否包含必填字段：

def validate_field_completeness(query, required_fields):
    collection = get_mongo_collection()
    invalid_docs = []
    for doc in collection.find(query):
        missing_fields = [field for field in required_fields if field not in doc]
        if missing_fields:
            invalid_docs.append({
                "_id": doc["_id"],
                "missing_fields": missing_fields
            })
    if invalid_docs:
        print("存在字段不完整的文档：")
        for doc in invalid_docs:
            print(f"文档ID: {doc['_id']}，缺失字段: {doc['missing_fields']}")
    else:
        print("所有匹配文档的字段均完整")
    return invalid_docs

# 调用示例：检查包含AcctEntxxxx的文档是否包含"UserName"、"Balance"字段
validate_field_completeness({"AcctEntxxxx": {"$exists": True}}, ["UserName", "Balance"])

4. 验证字段格式/类型正确性 #

检查目标字段的类型或格式是否符合预期：

def validate_field_format(query, field_name, expected_type=None, pattern=None):
    collection = get_mongo_collection()
    invalid_docs = []
    for doc in collection.find(query):
        field_value = doc.get(field_name)
        # 检查类型是否匹配
        if expected_type and not isinstance(field_value, expected_type):
            invalid_docs.append({
                "_id": doc["_id"],
                "error": f"字段{field_name}类型错误，预期{expected_type.__name__}，实际{type(field_value).__name__}"
            })
            continue
        # 检查格式是否符合正则规则（如果有）
        if pattern and not pattern.match(str(field_value)):
            invalid_docs.append({
                "_id": doc["_id"],
                "error": f"字段{field_name}格式不符合要求，值为{field_value}"
            })
    if invalid_docs:
        print("存在字段格式错误的文档：")
        for doc in invalid_docs:
            print(f"文档ID: {doc['_id']}，错误信息: {doc['error']}")
    else:
        print(f"所有匹配文档的{field_name}字段格式均符合要求")
    return invalid_docs

# 调用示例：检查AcctEntxxxx字段是否为字符串类型
import re
validate_field_format({"AcctEntxxxx": {"$exists": True}}, "AcctEntxxxx", expected_type=str)
# 检查AcctEntxxxx是否为字母数字组合
pattern = re.compile(r'^[A-Za-z0-9]+$')
validate_field_format({"AcctEntxxxx": {"$exists": True}}, "AcctEntxxxx", pattern=pattern)

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