如何在FastAPI中实现MongoDB事务接口？-MongoDB-PHP中文网

FastAPI中MongoDB事务需副本集支持，本地开发应通过Docker启动带--replSet的Mongo实例并正确配置连接字符串；事务必须在同步上下文中执行，需用run_in_executor封装，且每次请求应独立获取并清理session，异常时显式abort，事务结束后丢弃session。

如何在fastapi中实现mongodb事务接口？

FastAPI里直接调用`session.start_transaction()`会报错

因为MongoDB事务必须在副本集（replica set）或分片集群上运行，单节点mongod默认不支持事务。启动服务时如果没加--replSet参数，哪怕代码里写了start_transaction()，也会立刻抛出Transaction numbers are only allowed on a replica set member错误。

实操建议：

本地开发先用docker run -d -p 27017:27017 --name mongo-rs -e MONGO_INITDB_ROOT_USERNAME=admin -e MONGO_INITDB_ROOT_PASSWORD=pass mongo:6 --replSet rs0启动副本集
接着进容器执行rs.initiate({ _id: "rs0", members: [{ _id: 0, host: "localhost:27017" }] })
连接时必须用mongodb://admin:pass@localhost:27017/?replicaSet=rs0，缺replicaSet参数照样失败

`asyncio`下MongoDB事务不能跨`await`边界

MongoDB Python Driver 的事务上下文（with session.start_transaction():）不是异步的，但FastAPI路由是async def。如果你在await调用数据库操作中间插入session.commit_transaction()，会触发RuntimeError: Cannot enter context with async function。

正确做法是把所有涉及事务的操作包进一个同步函数里，再用loop.run_in_executor调用：

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=4)

async def transfer_money(session, from_id, to_id, amount):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        executor,
        _do_transfer_sync,
        session, from_id, to_id, amount
    )

def _do_transfer_sync(session, from_id, to_id, amount):
    with session.start_transaction():
        accounts = session.client.mydb.accounts
        accounts.update_one({"_id": from_id}, {"$inc": {"balance": -amount}})
        accounts.update_one({"_id": to_id}, {"$inc": {"balance": amount}})

注意：session对象不能被序列化或跨线程传递，必须在run_in_executor内部重新获取——但这里我们是从FastAPI依赖注入中传入已创建好的ClientSession，所以得确保它来自同一个MotorClient实例且未关闭。

事务失败后`session.abort_transaction()`不是自动的

Driver不会在异常时自动回滚，必须显式调用abort_transaction()，否则该session会卡在“in transaction”状态，后续操作全被阻塞，直到超时（默认60秒）才释放。

mongodb-query

用于调试和故障排查，查询MongoDB数据库。用户需要时可以： (1) 列出所有数据库； (2) 列出数据库中的集合； (3) 执行MongoDB查询或命令。

下载

安全写法必须带try/except/finally：

with session.start_transaction():
    try:
        coll1.update_one(..., session=session)
        coll2.insert_one(..., session=session)
        session.commit_transaction()
    except Exception:
        session.abort_transaction()
        raise

常见坑：

只写except没写finally，异常后漏掉abort，下次用同一session就卡死
在commit_transaction()后又抛异常，此时事务已提交，再abort会报InvalidOperation: Cannot abort transaction that has been committed
session对象不能复用：一次事务结束后，这个session必须丢弃，下次事务要用新client.start_session()

FastAPI依赖注入里怎么安全提供`ClientSession`

不能把session当全局变量或单例注入，否则并发请求会共用一个session，引发事务冲突或状态污染。

推荐方式是每次请求生成独立session，并在请求结束时清理：

from fastapi import Depends, Request

async def get_session(request: Request):
    client = request.app.state.mongo_client
    session = client.start_session()
    try:
        yield session
    finally:
        session.end_session()

@app.post("/transfer")
async def transfer(
    payload: TransferRequest,
    session: ClientSession = Depends(get_session)
):
    await transfer_money(session, payload.from_id, payload.to_id, payload.amount)

关键点：