必须用AsyncIOMotorClient而非PyMongo.MongoClient，因后者会阻塞事件循环；客户端须全局单例并在应用启停时通过on_event管理生命周期；数据库对象不可预固化collection引用，应按需获取或通过参数化Depends动态选择；聚合游标须显式to_list或async for消费；写操作需主动检查result.acknowledged；连接字符串末尾必须加?retryWrites=true。
直接结论：事件循环冲突几乎总是因为混用了同步 MongoClient 和异步 Motor 客户端，或者在非协程上下文中执行了 await 操作。
为什么 AsyncIOMotorClient 会报“no running event loop”？
这个错误不是 Motor 的问题，而是你试图在没有事件循环的线程或函数里 await 了异步操作。
-
app.on_event("startup")必须声明为async def，否则内部的await client.admin.command("ping")会直接崩溃 - 模块顶层写
client = AsyncIOMotorClient(...)是无效的——它不触发连接，且无法 await 初始化逻辑 - 用
threading.Thread或concurrent.futures.ProcessPoolExecutor手动启新线程时，该线程默认无事件循环，await就会失败
Depends 里新建 AsyncIOMotorClient 为什么不行？
每次请求都调 AsyncIOMotorClient()，等于每秒新建几十个连接池，不是“阻塞”，而是资源爆炸。
- 连接池失控 → 系统打开文件数超限，报
OSError: Too many open files - DNS 频繁解析 → 大量请求下
getaddrinfo成瓶颈 - 内存泄漏 → 每个
AsyncIOMotorClient实例带独立连接池和监控器，不释放就一直占着 - 正确做法是全局单例 +
@app.on_event("startup")创建，@app.on_event("shutdown")调client.close()
PyMongo.MongoClient 套 async/await 有用吗？
完全没用。PyMongo 是同步驱动，所有方法（如 find_one）本质是阻塞式系统调用，event loop 会被卡住。
- 哪怕写成
async def get_user(): return await asyncio.to_thread(pymongo_col.find_one, {...})，也只是把阻塞挪到线程池，没解决根本问题 - FastAPI 并发优势归零：一个请求卡住整个 worker，QPS 直接腰斩
- 必须换用
motor.motor_asyncio.AsyncIOMotorClient，它是基于asyncio和aiomongodb底层重写的
如何验证当前连接真正在异步运行？
最简单办法：加一行日志，在 startup 里打印 asyncio.get_running_loop() 的 id，并在某个路由 handler 里再打一次，两个 id 必须一致。
- 如果不一致，说明你无意中切到了另一个 loop（比如用
asyncio.new_event_loop()或nest_asyncio强行 patch） - 生产环境禁用
nest_asyncio.apply()—— 它掩盖问题，不解决设计缺陷 - 调试时可临时加
assert asyncio.get_event_loop() is asyncio.get_running_loop()，快速暴露 loop 切换点
真正容易被忽略的是：Motor 的“异步”只保证 I/O 不阻塞，但业务逻辑（比如大量 JSON 序列化、正则匹配、嵌套 for 循环）仍可能拖慢 event loop。别把 CPU 密集型操作塞进 async def 里——该丢给 asyncio.to_thread 就丢，该上 Celery 就上。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
