Skip to content

mrhasankoc/fastapi-tips

 
 

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

3 Commits
 
 

Repository files navigation

101 FastAPI Tips by The FastAPI Expert

This repository contains trips 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 sponsor me on GitHub to support my work. With your support, I will be able to create more content like this.

GitHub Sponsors

1. Install uvloop and httptools

By default, Uvicorn doesn't comes 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.

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:

import anyio
from contextlib import asynccontextmanager
from typing import Iterator

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> Iterator[None]:
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield

app = 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:

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

You can use the async for notation:

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    async for data in websocket.iter_text():
        await websocket.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.

from fastapi import FastAPI
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        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.

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


# Using TestClient
from starlette.testclient import TestClient

client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}

# Using AsyncClient
import anyio
from httpx import AsyncClient


async def main():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.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.

from contextlib import asynccontextmanager
from typing import AsyncIterator

import anyio
from asgi_lifespan import LifespanManager
from httpx import AsyncClient
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    print("Starting app")
    yield
    print("Stopping app")


app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


async def main():
    async with LifespanManager(app, lifespan) as manager:
        async with AsyncClient(app=manager.app) as client:
            response = await client.get("/")
            assert response.status_code == 200
            assert response.json() == {"Hello": "World"}


anyio.run(main)

Note

Consider supporting the creator of asgi-lifespan Florimond 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:

from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI, Request
from httpx import AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with AsyncClient(app=app) as client:
        app.state.client = client
        yield
        await client.aclose()


app = FastAPI()


@app.get("/")
async def read_root(request: Request):
    client = request.app.state.client
    response = await client.get("/")
    return response.json()

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

from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, TypedDict, cast

from fastapi import FastAPI, Request
from httpx import AsyncClient


class State(TypedDict):
    client: AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with AsyncClient(app=app) as client:
        yield {"client": client}


app = FastAPI()


@app.get("/")
async def read_root(request: Request) -> dict[str, Any]:
    client = cast(AsyncClient, request.state.client)
    response = await client.get("/")
    return response.json()

About

FastAPI Tips by The FastAPI Expert!

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published