Python motor vs pymongo 的异步演进

motor是基于asyncio重写的异步mongodb驱动，非pymongo简单封装；需用await调用、async for遍历、显式管理客户端生命周期，并注意pydantic校验前必须await查询结果。

motor 是 pymongo 的异步替代，但不是简单加个 async 就能跑

motor 不是 pymongo 的“异步版封装”，而是基于 asyncio 重写的驱动，底层用 asyncio.EventLoop 调度 I/O，pymongo 则完全阻塞。直接把 pymongo 代码里的 Collection.find() 换成 collection.find()（motor）会报错——motor 返回的是 AsyncIOMotorCursor，不是可迭代对象。

• 常见错误现象：TypeError: 'AsyncIOMotorCursor' Object is not iterable
• 必须显式调用 to_list()、next() 或用 async for 遍历
• 不支持同步调用：哪怕在 jupyter 里用 await 也要确保当前环境已启动 event loop
• pymongo 的 max_time_ms、skip、limit 等参数 motor 全部兼容，但行为一致不代表写法一致——比如 find_one() 在 motor 中仍是 await collection.find_one(...)，少个 await 就卡住不动

连接池和客户端生命周期必须手动管理

motor 的 AsyncIOMotorClient 不像某些 ORM 那样自动复用或懒初始化。它内部维护一个异步连接池，但不会自动关闭；如果每次请求都新建 client，会快速耗尽 socket 资源或触发 ConnectionAbortedError。

• 典型误用：在 fastapi 的依赖函数里每次 return 新的 AsyncIOMotorClient
• 正确做法：全局单例或通过 lifespan hook 初始化一次，在应用退出时调用 client.close()
• 注意 client.close() 是协程，得 await；不 await 只是标记关闭，实际连接可能还在
• 连接字符串里的 maxPoolSize 默认是 100，高并发场景下需结合 uvicorn worker 数调整，否则容易出现 ServerSelectionTimeoutError

聚合管道里不能混用同步/异步 cursor

motor 的 aggregate() 返回 AsyncIOMotorCommandCursor，它不支持 list(cursor) 或直接 for doc in cursor。有人试图用 loop.run_until_complete() 强制同步执行，结果卡死或抛出 RuntimeError: this event loop is already running。

• 错误写法：list(collection.aggregate([...])) → 类型错误 + 死锁
• 正确路径只有两条：用 await cursor.to_list(Length=100)（适合小结果集），或 async for doc in cursor（流式处理大结果）
• 聚合阶段如 $lookup 或 $facet 不影响 motor 的使用方式，但要注意内存：to_list() 会一次性加载所有文档到内存，而 async for 是逐条拉取
• 如果管道里用了自定义 JavaScript（$where），motor 和 pymongo 行为一致，但 js 执行本身仍由 mongodb 服务端完成，motor 只负责收发——这点常被误认为“JS 也会变异步”

Pydantic v2 + motor 的类型提示容易漏掉 await

用 Pydantic v2 做响应模型时，很多人把 motor 查询结果直接传给 model_validate()，忘了结果还没 await。例如 User.model_validate(collection.find_one(...)) 实际传入的是一个 coroutine 对象，不是 dict，导致 ValidationError 提示 “expected dict”。

立即学习“Python免费学习笔记（深入）”；

• 最常见坑：find_one()、insert_one()、update_one() 全部返回协程，必须 await
• insert 后想取 inserted_id？得写 (await collection.insert_one(...)).inserted_id
• Pydantic 的 model_validate() 不接受 coroutine，model_validate_json() 也不行——没数据，只有 await 后的 dict 或 bytes 才行
• ide 很难静态检查这种漏 await，建议在 mypy 配置中启用 disallow_any_unimported 并配合 asyncio 插件，但最靠谱的还是单元测试里加 asyncio.run() 跑通流程

motor 的异步不是语法糖，是 I/O 调度模型的切换。最容易被忽略的是：你以为只是加个 await，其实是在重新组织整个请求生命周期——从连接创建、查询发起、结果消费，到错误传播链，每一步的等待点都得对齐 event loop。写错一个 await，整条链就断在那儿，还很难 debug。