如何为PyMongo正确添加类型注解？多集合场景下的类型安全方案问询

1. 利用PyMongo内置泛型获取指定类型集合（推荐） #

如果你的PyMongo版本在4.3及以上，直接使用Database.get_collection的泛型参数指定文档类型，无需手动cast，这是最简洁的方式：

from typing import Any, TypedDict
from pymongo import MongoClient
from pymongo.collection import Collection
from app import app
from flask import jsonify, make_response

class Movie(TypedDict):
    title: str
    genres: list[str]

def get_db_client() -> MongoClient[Any]:
    conn_string = str(app.config['MONGO_URI'])
    return MongoClient(conn_string)

@app.route('/test')
def models():
    client = get_db_client()
    db = client['sample_mflix']
    # 直接通过泛型参数指定集合的文档类型
    coll = db.get_collection[Movie]("movies")
    
    genres = list(coll.aggregate([
        {"$unwind": "$genres"},
        {"$group": {"_id": None, "genres": {"$addToSet": "$genres"}}}
    ]))[0]['genres']

    return make_response(jsonify(genres), 200)

这种方式的优势：

• 复用单一MongoClient实例（MongoClient内置连接池，线程安全，无需重复创建）
• 直接获得类型安全的集合对象，无需额外转换逻辑
• 支持任意集合的类型标注，适配多业务场景

2. 封装泛型工具函数统一处理类型转换 #

如果需要兼容旧版本PyMongo，或者想把类型转换逻辑集中管理，可以封装一个泛型函数：

from typing import Any, TypedDict, TypeVar, cast
from pymongo import MongoClient
from pymongo.collection import Collection
from app import app
from flask import jsonify, make_response

T = TypeVar('T', bound=TypedDict)

class Movie(TypedDict):
    title: str
    genres: list[str]

class User(TypedDict):
    username: str
    email: str

def get_db_client() -> MongoClient[Any]:
    conn_string = str(app.config['MONGO_URI'])
    return MongoClient(conn_string)

def get_typed_collection(db_name: str, coll_name: str) -> Collection[T]:
    client = get_db_client()
    db = client[db_name]
    return cast(Collection[T], db[coll_name])

@app.route('/test')
def models():
    # 调用时指定泛型参数，获得类型安全的集合
    coll = get_typed_collection[Movie]("sample_mflix", "movies")
    genres = list(coll.aggregate([
        {"$unwind": "$genres"},
        {"$group": {"_id": None, "genres": {"$addToSet": "$genres"}}}
    ]))[0]['genres']
    return make_response(jsonify(genres), 200)

@app.route('/users')
def get_users():
    # 适配其他集合类型
    user_coll = get_typed_collection[User]("sample_mflix", "users")
    users = list(user_coll.find({}, {"username": 1}))
    return make_response(jsonify(users), 200)

这种方式的优势：

• 统一管理类型转换逻辑，避免业务代码中重复写cast
• 同样复用单一MongoClient实例，资源利用率高
• 支持多集合的类型标注，扩展性强

3. 不推荐的方案 #

• 为每个集合创建独立客户端：完全没必要，MongoClient本身是线程安全的连接池容器，重复创建会浪费资源，且没有类型安全上的额外收益。
• 业务代码中大量使用cast：会导致代码冗余，维护性差，仅适合临时小范围场景使用。

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