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()跑通流程
