fastapi: How does one setup a global timeout to all requests?

This is good to return an error message to the user in case of timeout, but is there a way to actually kill the request at the same time so it doesn’t keep using resources?

Hey @PLNech

I have implemented and tested the middleware and it seems to be working fine for me. Here is my code

import asyncio
import time


import pytest

from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.responses import JSONResponse
from httpx import AsyncClient
from starlette.status import HTTP_504_GATEWAY_TIMEOUT

REQUEST_TIMEOUT_ERROR = 1  # Threshold

app = FastAPI() # Fake app

# Creating a test path
@app.get("/test_path")
async def route_for_test(sleep_time: float) -> None:
    await asyncio.sleep(sleep_time)

# Adding a middleware returning a 504 error if the request processing time is above a certain threshold
@app.middleware("http")
async def timeout_middleware(request: Request, call_next):
    try:
        start_time = time.time()
        return await asyncio.wait_for(call_next(request), timeout=REQUEST_TIMEOUT_ERROR)

    except asyncio.TimeoutError:
        process_time = time.time() - start_time
        return JSONResponse({'detail': 'Request processing time excedeed limit',
                             'processing_time': process_time},
                            status_code=HTTP_504_GATEWAY_TIMEOUT)

# Testing wether or not the middleware triggers
@pytest.mark.asyncio
async def test_504_error_triggers():
    # Creating an asynchronous client to test our asynchronous function
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/test_path?sleep_time=3")
    content = eval(response.content.decode())
    assert response.status_code == HTTP_504_GATEWAY_TIMEOUT
    assert content['processing_time'] < 1.1

# Testing middleware's consistency for requests having a processing time close to the threshold 
@pytest.mark.asyncio
async def test_504_error_consistency():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        errors = 0
        sleep_time = REQUEST_TIMEOUT_ERROR*0.9
        for i in range(100):
            response = await ac.get("/test_path?sleep_time={}".format(sleep_time))
            if response.status_code == HTTP_504_GATEWAY_TIMEOUT:
                errors += 1
        assert errors == 0

# Testing middleware's precision
# ie : Testing if it triggers when it should not and vice versa
@pytest.mark.asyncio
async def test_504_error_precision():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        should_trigger = []
        should_pass = []
        have_triggered = []
        have_passed = []
        for i in range(200):
            sleep_time = 2 * REQUEST_TIMEOUT_ERROR * random.random()
            if sleep_time < 1.1:
                should_pass.append(i)
            else:
                should_trigger.append(i)
            response = await ac.get("/test_path?sleep_time={}".format(sleep_time))
            if response.status_code == HTTP_504_GATEWAY_TIMEOUT:
                have_triggered.append(i)
            else:
                have_passed.append(i)
        assert should_trigger == have_triggered

I created three tests, the first one is designed to see wether or not the middleware actually does its job. The second one is just there to check if there is any consistency problem with a single request. The third one is here to check if I ran into the same issue raised in the thread I mentioned.

As far as I am concerned the first two tests passed without a problem. However the third one failed. There are requests that have triggered when they should not :

E           AssertionError: assert [3, 7, 10, 11, 12, 14, ...] == [3, 7, 8, 10, 11, 12, ...]
E             At index 2 diff: 10 != 8
E             Right contains 11 more items, first extra item: 165

This is the issue mentioned in the thread. I’ll downgrade to starlette 0.13.2 and see if the test pass.

I might have made some mistakes or overlooked some things so I you ever have the chance to do some tests on your end let me know.

Cheers !

Note : I wrote assert content['processing_time'] < 1.1 and not assert content['processing_time'] < 1 because the time I am monitoring isn’t really the time it takes for python to execute the function (time to execute asyncio.wait_for and catching the exception I guess) . I do not know the convention in this case.

@thomas-maschler: thanks for the advice. Unfortunately I’ve tried using Gunicorn’s timeout, but it triggers a full restart of the app, disrupting other users of the service (e.g. by unloading their models from memory). What I’m trying to achieve is rather to enforce a timeout on individual requests, without affecting any other work handled by this worker.

Workaround: I’ve created a decorator to use in the endpoints you want to raise a response 504: (place it in a file named abort_after.py)

import functools
import signal
import sys

from fastapi.responses import JSONResponse
from starlette import status


class TimeOutException(Exception):
    """It took longer than expected"""


def abort_after(max_execution_time):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            def handle_timeout(signum, frame):
                raise TimeOutException(f"Function execution took longer than {max_execution_time}s and was terminated")
            if sys.platform == 'win32':
                print("Won't be stopped in windows!")
            else:
                signal.signal(signal.SIGALRM, handle_timeout)
                signal.alarm(max_execution_time)
            result = func(*args, **kwargs)
            if sys.platform != 'win32':
                signal.alarm(0)
            return result
        return wrapper
    return decorator


def timeout_response() -> JSONResponse:
    return JSONResponse(
        {
            'detail': 'Request processing time excedeed limit',
        },
        status_code=status.HTTP_504_GATEWAY_TIMEOUT,
    )

Then you can use it in your endpoint:

import time
from fastapi import APIRouter
from abort_after import abort_after, TimeOutException, timeout_response

router = APIRouter()


@router.post(f"{URL_prefix}/test",
             tags=['Test'],
             )
async def test():
    try:
        long_func(60)
    except TimeOutException:
        return timeout_response()
    return {'Test': 'ok'}


@abort_after(5)
def long_func(seconds: int) -> None:
    time.sleep(seconds)

Hi @ZionStage, thanks for your message! I haven’t found a workaround for now. Looking forward to continuing this conversation with you as we move forward on this topic 😃

Hi @PLNech

I am developing my own API using FastAPI and ran into the same “problem” as I am trying to add a global timeout to all my requests.

I am still new to fastapi but from what I understand I believe the “fastapi” way to do so would be to use a middleware as they are designed to be ran at every request by nature. As I searched on how to do so I found this gitter community thread and thought it could maybe help you.

I am going to implement both your solution and the middleware based one and see which one I prefer and works best. Also note that there seems to be a problem with starlette 0.13.3 and higher so keep that in mind.

Also if you found a workaround by now I am more than interested.

Hope it helped you a bit