必须用AsyncIOMotorClient而非PyMongo.MongoClient,因后者会阻塞事件循环;客户端须全局单例并绑定startup/shutdown生命周期;数据库和collection引用不可固化,应按需获取;聚合游标须显式to_list或async for消费;连接字符串末尾必须加?retryWrites=true。

必须用 AsyncIOMotorClient,不能用 PyMongo.MongoClient——后者哪怕套 async/await 也会阻塞事件循环,FastAPI 的并发优势直接归零。
连接客户端必须全局单例,且绑定应用启停生命周期
Motor 客户端本身线程安全、协程安全,但频繁新建 AsyncIOMotorClient 实例会导致连接池失控、内存泄漏、DNS 频繁解析。
- ✅ 正确做法:在
@app.on_event("startup")中创建 client,在@app.on_event("shutdown")中调用client.close() - ❌ 错误写法:
Depends函数里每次请求都AsyncIOMotorClient();或在模块顶层直接client = AsyncIOMotorClient(...)(此时无法await初始化) - ⚠️ 注意:不要在
startup里await client.admin.command("ping")——必须包在异步函数内,否则报RuntimeError: no running event loop
数据库和集合不能提前固化引用
像 users_col = db["users"] 这种赋值一旦写成全局变量,多租户、多数据库场景下极易串库。Motor 的 collection 对象不保存状态,但引用会固定到某个 db 实例。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成,提供 184 种工具和 200 多条命令,用于安全分析和代码质量...
- ✅ 推荐模式:每个路由 handler 或依赖函数中按需获取
db["collection_name"],或封装为带参数的Depends(如def get_collection(name: str)) - ✅ 更安全的抽象:用
Depends返回AsyncIOMotorDatabase,让业务层自己选 collection,避免预绑定 - ⚠️ 常见坑:把
db["logs"]赋给全局变量后,在不同请求中混用,导致日志写进错误库或权限越界
聚合查询游标必须显式消费,不能直接 for 循环
collection.aggregate(pipeline) 返回的是 AsyncIOMotorCommandCursor,不是 list,也不是同步可迭代对象。直接 for doc in cursor: 会报 TypeError 或静默失败。
- ✅ 拿全部结果:用
await cursor.to_list(length=None)(小数据量)或await cursor.to_list(length=1000)(防 OOM) - ✅ 流式处理大数据:用
async for doc in cursor:,配合yield做 Server-Sent Events 或分块响应 - ⚠️ 性能陷阱:管道里含
$lookup却没在被关联集合上建索引,Motor不会预警,查询延迟飙升且拖慢整个服务
连接字符串末尾必须加 ?retryWrites=true,否则某些写操作可能因网络抖动失败且不重试;maxPoolSize 和 minPoolSize 参数要根据实际并发量调,别盲目设成 100——连接数远超 MongoDB 默认 maxIncomingConnections(通常 65536)时,反而触发拒绝连接。

















