FastAPI中Pymongo单数据库连接实现正确性及最佳实践问询

你的单连接实现是否正确？ #

思路是对的，但有几个关键细节要落地：

• 必须确保Database类是单例，比如通过重写__new__控制实例化，或者在FastAPI启动时只创建一次实例并全局注入，否则还是会出现多连接
• 关闭逻辑要绑定到FastAPI的 shutdown 事件，比如：

  @app.on_event("shutdown")
  async def close_db_connection():
      await your_db_instance.disconnect()
  

• 如果用的是同步数据库驱动，单连接会在高并发场景下导致请求阻塞——因为同一时间只能有一个请求用这个连接处理数据

是否需要避免多连接？ #

完全没必要，反而推荐用连接池管理多连接：

• FastAPI本身支持高并发，单连接会成为性能瓶颈，所有请求都得排队等连接释放
• 数据库原生就支持多连接，连接池能复用已有连接，避免频繁创建/销毁连接的开销
• 主流ORM或数据库驱动都自带成熟的连接池实现，配置成本极低

生产环境最佳实践 #

1. SQLAlchemy 方案（最常用） #

不管同步还是异步场景，SQLAlchemy都能自动管理连接池，无需手动操心：

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base

# 同步数据库URL，异步用create_async_engine
DATABASE_URL = "mysql+pymysql://user:pass@localhost/db"

engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 每个请求生成一个会话，自动从连接池取/放连接
def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

路由里直接用Depends(get_db)注入会话即可，FastAPI会自动处理每个请求的会话生命周期。

2. 原生驱动+连接池 #

如果不用ORM，比如直接用aiopg（异步PostgreSQL），可以手动配置连接池：

# database.py
import aiopg
from typing import AsyncGenerator

async def create_db_pool():
    return await aiopg.create_pool(
        host="localhost",
        user="admin",
        password="123456",
        database="test",
        minsize=5,
        maxsize=15
    )

async def get_db_connection(pool=Depends(create_db_pool)) -> AsyncGenerator[aiopg.Connection, None]:
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

然后在main.py里处理连接池关闭：

@app.on_event("shutdown")
async def shutdown_pool(pool=Depends(create_db_pool)):
    pool.close()
    await pool.wait_closed()

最后总结 #

• 单连接只适合低并发的测试场景，生产环境别用
• 优先用连接池+请求级连接/会话的模式，既保证并发性能，又能自动管理资源
• FastAPI的Depends+事件钩子是管理数据库资源的标准姿势，比手动维护全局单连接靠谱得多

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