101 FastAPI Tips by The FastAPI Expert

This repository contains tips and tricks for FastAPI. If you have any tip that you believe is useful, feel free to open an issue or a pull request.

Consider sponsoring me on GitHub to support my work. With your support, I will be able to create more content like this.

Tip

Remember to watch this repository to receive notifications about new tips.

1. Install uvloop and httptools

By default, Uvicorn doesn't come with uvloop and httptools which are faster than the default asyncio event loop and HTTP parser. You can install them using the following command:

pip install uvloop httptools

Uvicorn will automatically use them if they are installed in your environment.

Warning

uvloop can't be installed on Windows. If you use Windows locally, but Linux on production, you can use an environment marker to not install uvloop on Windows e.g. uvloop; sys_platform != 'win32'.

2. Be careful with non-async functions

There's a performance penalty when you use non-async functions in FastAPI. So, always prefer to use async functions. The penalty comes from the fact that FastAPI will call run_in_threadpool, which will run the function using a thread pool.

Note

Internally, run_in_threadpool will use anyio.to_thread.run_sync to run the function in a thread pool.

Tip

There are only 40 threads available in the thread pool. If you use all of them, your application will be blocked.

To change the number of threads available, you can use the following code:

importanyiofromcontextlibimportasynccontextmanagerfromtypingimportIteratorfromfastapiimportFastAPI@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->Iterator[None]:
limiter=anyio.to_thread.current_default_thread_limiter()
limiter.total_tokens=100yieldapp=FastAPI(lifespan=lifespan)

You can read more about it on AnyIO's documentation.

3. Use async for instead of while True on WebSocket

Most of the examples you will find on the internet use while True to read messages from the WebSocket.

I believe the uglier notation is used mainly because the Starlette documentation didn't show the async for notation for a long time.

Instead of using the while True:

fromfastapiimportFastAPIfromstarlette.websocketsimportWebSocketapp=FastAPI()
@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket: WebSocket) ->None:
awaitwebsocket.accept()
whileTrue:
data=awaitwebsocket.receive_text()
awaitwebsocket.send_text(f"Message text was: {data}")

You can use the async for notation:

fromfastapiimportFastAPIfromstarlette.websocketsimportWebSocketapp=FastAPI()
@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket: WebSocket) ->None:
awaitwebsocket.accept()
asyncfordatainwebsocket.iter_text():
awaitwebsocket.send_text(f"Message text was: {data}")

You can read more about it on the Starlette documentation.

4. Ignore the WebSocketDisconnect exception

If you are using the while True notation, you will need to catch the WebSocketDisconnect. The async for notation will catch it for you.

fromfastapiimportFastAPIfromstarlette.websocketsimportWebSocket, WebSocketDisconnectapp=FastAPI()
@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket: WebSocket) ->None:
awaitwebsocket.accept()
try:
whileTrue:
data=awaitwebsocket.receive_text()
awaitwebsocket.send_text(f"Message text was: {data}")
exceptWebSocketDisconnect:
pass

If you need to release resources when the WebSocket is disconnected, you can use that exception to do it.

If you are using an older FastAPI version, only the receive methods will raise the WebSocketDisconnect exception. The send methods will not raise it. In the latest versions, all methods will raise it. In that case, you'll need to add the send methods inside the try block.

5. Use HTTPX's AsyncClient instead of TestClient

Since you are using async functions in your application, it will be easier to use HTTPX's AsyncClient instead of Starlette's TestClient.

fromfastapiimportFastAPIapp=FastAPI()
@app.get("/")asyncdefread_root():
return {"Hello": "World"}
# Using TestClientfromstarlette.testclientimportTestClientclient=TestClient(app)
response=client.get("/")
assertresponse.status_code==200assertresponse.json() == {"Hello": "World"}
# Using AsyncClientimportanyiofromhttpximportAsyncClient, ASGITransportasyncdefmain():
asyncwithAsyncClient(transport=ASGITransport(app=app), base_url="http://test") asclient:
response=awaitclient.get("/")
assertresponse.status_code==200assertresponse.json() == {"Hello": "World"}
anyio.run(main)

If you are using lifespan events (on_startup, on_shutdown or the lifespan parameter), you can use the asgi-lifespan package to run those events.

fromcontextlibimportasynccontextmanagerfromtypingimportAsyncIteratorimportanyiofromasgi_lifespanimportLifespanManagerfromhttpximportAsyncClient, ASGITransportfromfastapiimportFastAPI@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->AsyncIterator[None]:
print("Starting app")
yieldprint("Stopping app")
app=FastAPI(lifespan=lifespan)
@app.get("/")asyncdefread_root():
return {"Hello": "World"}
asyncdefmain():
asyncwithLifespanManager(app) asmanager:
asyncwithAsyncClient(transport=ASGITransport(app=manager.app)) asclient:
response=awaitclient.get("/")
assertresponse.status_code==200assertresponse.json() == {"Hello": "World"}
anyio.run(main)

Note

Consider supporting the creator of asgi-lifespanFlorimond Manca via GitHub Sponsors.

6. Use Lifespan State instead of app.state

Since not long ago, FastAPI supports the lifespan state, which defines a standard way to manage objects that need to be created at startup, and need to be used in the request-response cycle.

The app.state is not recommended to be used anymore. You should use the lifespan state instead.

Using the app.state, you'd do something like this:

fromcontextlibimportasynccontextmanagerfromtypingimportAsyncIteratorfromfastapiimportFastAPI, RequestfromhttpximportAsyncClient@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->AsyncIterator[None]:
asyncwithAsyncClient(app=app) asclient:
app.state.client=clientyieldapp=FastAPI(lifespan=lifespan)
@app.get("/")asyncdefread_root(request: Request):
client=request.app.state.clientresponse=awaitclient.get("/")
returnresponse.json()

Using the lifespan state, you'd do something like this:

fromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerfromtypingimportAny, TypedDict, castfromfastapiimportFastAPI, RequestfromhttpximportAsyncClientclassState(TypedDict):
client: AsyncClient@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->AsyncIterator[State]:
asyncwithAsyncClient(app=app) asclient:
yield {"client": client}
app=FastAPI(lifespan=lifespan)
@app.get("/")asyncdefread_root(request: Request) ->dict[str, Any]:
client=cast(AsyncClient, request.state.client)
response=awaitclient.get("/")
returnresponse.json()

7. Enable AsyncIO debug mode

If you want to find the endpoints that are blocking the event loop, you can enable the AsyncIO debug mode.

When you enable it, Python will print a warning message when a task takes more than 100ms to execute.

Run the following code with PYTHONASYNCIODEBUG=1 python main.py:

importosimporttimeimportuvicornfromfastapiimportFastAPIapp=FastAPI()
@app.get("/")asyncdefread_root():
time.sleep(1) # Blocking callreturn {"Hello": "World"}
if__name__=="__main__":
uvicorn.run(app, loop="uvloop")

If you call the endpoint, you will see the following message:

INFO: Started server process [19319]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: 127.0.0.1:50036 - "GET / HTTP/1.1" 200 OK
Executing <Task finished name='Task-3' coro=<RequestResponseCycle.run_asgi() done, defined at /uvicorn/uvicorn/protocols/http/httptools_impl.py:408> result=None created at /uvicorn/uvicorn/protocols/http/httptools_impl.py:291> took 1.009 seconds

You can read more about it on the official documentation.

8. Implement a Pure ASGI Middleware instead of BaseHTTPMiddleware

The BaseHTTPMiddleware is the simplest way to create a middleware in FastAPI.

Note

The @app.middleware("http") decorator is a wrapper around the BaseHTTPMiddleware.

There were some issues with the BaseHTTPMiddleware, but most of the issues were fixed in the latest versions. That said, there's still a performance penalty when using it.

To avoid the performance penalty, you can implement a Pure ASGI middleware. The downside is that it's more complex to implement.

Check the Starlette's documentation to learn how to implement a Pure ASGI middleware.

9. Your dependencies may be running on threads

If the function is non-async and you use it as a dependency, it will run in a thread.

In the following example, the http_client function will run in a thread:

fromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerfromhttpximportAsyncClientfromfastapiimportFastAPI, Request, Depends@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->AsyncIterator[dict[str, AsyncClient]]:
asyncwithAsyncClient() asclient:
yield {"client": client}
app=FastAPI(lifespan=lifespan)
defhttp_client(request: Request) ->AsyncClient:
returnrequest.state.client@app.get("/")asyncdefread_root(client: AsyncClient=Depends(http_client)):
returnawaitclient.get("/")

To run in the event loop, you need to make the function async:

# ...asyncdefhttp_client(request: Request) ->AsyncClient:
returnrequest.state.client# ...

As an exercise for the reader, let's learn a bit more about how to check the running threads.

You can run the following with python main.py:

fromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerimportanyiofromanyio.to_threadimportcurrent_default_thread_limiterfromhttpximportAsyncClientfromfastapiimportFastAPI, Request, Depends@asynccontextmanagerasyncdeflifespan(app: FastAPI) ->AsyncIterator[dict[str, AsyncClient]]:
asyncwithAsyncClient() asclient:
yield {"client": client}
app=FastAPI(lifespan=lifespan)
# Change this function to be async, and rerun this application.defhttp_client(request: Request) ->AsyncClient:
returnrequest.state.client@app.get("/")asyncdefread_root(client: AsyncClient=Depends(http_client)): ...
asyncdefmonitor_thread_limiter():
limiter=current_default_thread_limiter()
threads_in_use=limiter.borrowed_tokenswhileTrue:
ifthreads_in_use!=limiter.borrowed_tokens:
print(f"Threads in use: {limiter.borrowed_tokens}")
threads_in_use=limiter.borrowed_tokensawaitanyio.sleep(0)
if__name__=="__main__":
importuvicornconfig=uvicorn.Config(app="main:app")
server=uvicorn.Server(config)
asyncdefmain():
asyncwithanyio.create_task_group() astg:
tg.start_soon(monitor_thread_limiter)
awaitserver.serve()
anyio.run(main)

If you call the endpoint, you will see the following message:

❯ python main.py
INFO: Started server process [23966]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Threads in use: 1
INFO: 127.0.0.1:57848 - "GET / HTTP/1.1" 200 OK
Threads in use: 0

Replace the def http_client with async def http_client and rerun the application. You will not see the message Threads in use: 1, because the function is running in the event loop.

Tip

You can use the FastAPI Dependency package that I've built to make it explicit when a dependency should run in a thread.

10. Use pytest.mark.anyio instead of pytest.mark.asyncio

You already have anyio installed, since it's a dependency of Starlette. Which means, you can use pytest.mark.anyio instead of pytest.mark.asyncio.

importpytest@pytest.mark.anyioasyncdeftest_async_function(): ...

By default, anyio runs every test that has the marker twice, once with trio and another time with asyncio. You probably want to restrict that by using either one or the other, in case you are testing an application, and not a package:

importpytest@pytest.fixturedefanyio_backend():
return"asyncio"# or "trio"

You can read more about it on the anyio documentation.