I am using nsidnev/fastapi-realworld-example-app.
I need to apply transaction logic to this project.
In one API, I am calling a lot of methods from repositories and doing updating, inserting and deleting operations in many tables. If there is an exception in any of these operations, how can I roll back changes? (Or if everything is correct then commit.)
This is a Python function. It will be called by FastAPI whenever it receives a request to the URL " / " using a GET operation. In this case, it is an async function.
And there are dozens of alternatives, all based on OpenAPI. You could easily add any of those alternatives to your application built with FastAPI. You could also use it to generate code automatically, for clients that communicate with your API. For example, frontend, mobile or IoT applications.
Recap, step by step. 1 Step 1: import FastAPI. from fastapi import FastAPI app = FastAPI() @app.get("/") async def root(): return {"message": "Hello World"} FastAPI is a ... 2 Step 2: create a FastAPI "instance". 3 Step 3: create a path operation. 4 Step 4: define the path operation function. 5 Step 5: return the content.
So, in OpenAPI, each of the HTTP methods is called an "operation". We are going to call them " operations " too. The @app.get ("/") tells FastAPI that the function right below is in charge of handling requests that go to: That @something syntax in Python is called a "decorator".
nsidnev/fastapi-realworld-example-app is using asyncpg.
There are two ways to use Transactions.
async with
statementasync 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
start
, rollback
, commit
statementstx = 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
in routesInject 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:
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With