如何在FastAPI RealWorld示例应用程序中应用事务逻辑

Question

我正在使用nsidnev/fastapi现实世界示例应用程序。

我需要将事务逻辑应用于这个项目。

在一个API中，我从存储库调用了许多方法，并在许多表中执行更新、插入和删除操作。如果这些操作中有异常，我如何回滚更改？(或者，如果一切都正确，那么提交。(

Answer 1

nsidnev/fastapi现实世界示例应用程序正在使用asyncpg。

使用事务有两种方法。

1.async with语句

async with conn.transaction():
await repo_one.update_one(...)
await repo_two.insert_two(...)
await repo_three.delete_three(...)
# This automatically rolls back the transaction:
raise Exception

2.start、rollback、commit语句

tx = conn.transaction()
await tx.start()
try:
await repo_one.update_one(...)
await repo_two.insert_two(...)
await repo_three.delete_three(...)
except:
await tx.rollback()
raise
else:
await tx.commit()

---

在路由中获取连接conn

注入conn: Connection = Depends(_get_connection_from_pool)。

from asyncpg.connection import Connection
from fastapi import Depends
from app.api.dependencies.database import _get_connection_from_pool

@router.post(
...
)
async def create_new_article(
...
conn: Connection = Depends(_get_connection_from_pool),  # Add this
) -> ArticleInResponse: