How to ensure that a block of code can be executed only by one request at a time in FastAPI

[mailtodanish]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

no code

Description

How to ensure that a block of code can be executed only by one request at a time in FAST API

I am implementing post and calling some external tool but it takes one request at a time.

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.70.0

Python Version

Python 3.8.5

Additional Context

No response

[MatthijsKok]

You can use FastAPI's Background Tasks together with Python's thread locks to achieve this.
See also this stackoverflow thread

[mailtodanish]

@MatthijsKok I need response from block of code and will send back to request. I don't think background task will return any response.

[MatthijsKok]

Something like this?

api.lock = False

@api.get("/task")
def do_task():
    if not myapp.lock:
        try:
            myapp.lock = True
            #do your external request here synchronously
            return "Task finished"
        finally:
            myapp.lock = False
    else:
        raise HTTPException(status_code=400,detail="Task already running.")

[mailtodanish]

thanks @MatthijsKok api.lock will be shared in all requests? any suggest how to wait if already locked? because every request should execute block of code.
Transaction will be created in external tool but external tool accepts one request at a time. Fast API is to create transaction in external tool using CLI.

[MatthijsKok]

What might be the best solution is to make sure that nothing is async in your api route, and then run the server with only one worker. That way you ensure that requests wait for eachother.
uvicorn --workers 1

[ycd]

There is no better way to solve this problem without using a custom router.

import asyncio
import time
from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute

from typing import Callable


class CustomRoute(APIRoute):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.lock = asyncio.Lock()

    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            await self.lock.acquire()
            response: Response = await original_route_handler(request)
            self.lock.release()
            return response

        return custom_route_handler


app = FastAPI()
app.router.route_class = CustomRoute


@app.get("/hello")
async def hello():
    time.sleep(5)
    return {"hello": "world"}

Even if you run this with uvicorn main:app --workers 4 this will work just fine.

[mailtodanish]

@ycd please explain the approach in detail. Thanks

[ycd]

Read the documentation: https://fastapi.tiangolo.com/advanced/custom-request-and-route/