Tips and Tricks

FastAPI

Mount a Flask or Django app inside a FastAPI application


FastAPI tip:

You can use WSGIMiddleware to mount WSGI applications (like Flask and Django) to your FastAPI API.

https://fastapi.tiangolo.com/advanced/wsgi/

👇

from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
from flask import Flask, escape, request

flask_app = Flask(__name__)


@flask_app.route("/")
def flask_main():
    name = request.args.get("name", "World")
    return f"Hello, {escape(name)} from Flask!"


app = FastAPI()


@app.get("/v2")
def read_main():
    return {"message": "Hello World"}


app.mount("/v1", WSGIMiddleware(flask_app))

FastAPI - disable OpenAPI docs


FastAPI tip:

You can disable OpenAPI docs by setting openapi_url to an empty string.

https://fastapi.tiangolo.com/advanced/conditional-openapi/#conditional-openapi-from-settings-and-env-vars

👇

from fastapi import FastAPI
from pydantic import BaseSettings


class Settings(BaseSettings):
    openapi_url: str = ""


settings = Settings()

app = FastAPI(openapi_url=settings.openapi_url)


@app.get("/")
def root():
    return {"message": "Hello World"}

FastAPI - custom Request and APIRoute class


FastAPI tip:

You can implement custom Request and APIRoute classes.

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

For example, to manipulate the request body before it's processed by your application👇

import gzip
from typing import Callable, List

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


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

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


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


@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body(...)):
    return {"sum": sum(numbers)}

FastAPI - GraphQL with Strawberry


FastAPI tip:

You can use Strawberry to build a GraphQL API with FastAPI.

🍓

https://fastapi.tiangolo.com/fr/advanced/graphql/#graphql-with-strawberry

👇

import strawberry

from fastapi import FastAPI
from strawberry.asgi import GraphQL


@strawberry.type
class User:
    name: str
    age: int


@strawberry.type
class Query:
    @strawberry.field
    def user(self) -> User:
        return User(name="Patrick", age=100)


schema = strawberry.Schema(query=Query)


graphql_app = GraphQL(schema)

app = FastAPI()
app.add_route("/graphql", graphql_app)
app.add_websocket_route("/graphql", graphql_app)

FastAPI - Templates with Jinja2


FastAPI tip:

You can use Jinja2 as a template engine to serve HTML responses from your FastAPI application.

👇

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")


templates = Jinja2Templates(directory="templates")


@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(request: Request, id: str):
    return templates.TemplateResponse("item.html", {"request": request, "id": id})

FastAPI Sub Applications


FastAPI tip:

You can use sub-applications when you need two separate OpenAPI schemas and Swagger UIs.

https://fastapi.tiangolo.com/advanced/sub-applications/

You can mount one or many sub-applications.

👇

from fastapi import FastAPI


app = FastAPI()


@app.get("/app")
def read_main():
    return {"message": "Hello World from main app"}


subapi = FastAPI()


@subapi.get("/sub")
def read_sub():
    return {"message": "Hello World from sub API"}


app.mount("/subapi", subapi)

FastAPI shutdown events


FastAPI tip:

You can register functions to run before the application shutdown using @app.on_event("shutdown").

https://fastapi.tiangolo.com/advanced/events/#shutdown-event

For example. to send a message to an SNS topic where you track your app health 👇

import boto3
from fastapi import FastAPI


app = FastAPI()


@app.on_event("shutdown")
def publish_message():
    client = boto3.client('sns')
    client.publish(
        TopicArn='arn:aws:sns:us-east-1:12345678910112:application-health',
        Message='Application is shutting down',
    )


@app.get("/ping")
async def ping():
    return {"message": "pong"}

FastAPI startup events


FastAPI tip:

You can register functions to run before the application start using @app.on_event("startup").

https://fastapi.tiangolo.com/advanced/events/#startup-event

For example, to send a message to an AWS SNS topic where you track your app health:

import boto3
from fastapi import FastAPI


app = FastAPI()


@app.on_event("startup")
def publish_message():
    client = boto3.client('sns')
    client.publish(
        TopicArn='arn:aws:sns:us-east-1:12345678910112:application-health',
        Message='Application is starting',
    )


@app.get("/ping")
async def ping():
    return {"message": "pong"}

FastAPI WebSockets


FastAPI tip:

You can easily add WebSockets to your app with @app.websocket().

https://fastapi.tiangolo.com/advanced/websockets/

👇

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


@app.get("/")
async def get():
    return HTMLResponse(html)


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

FastAPI - Using alias parameters to map fields from request to view arguments


FastAPI tip:

You can use aliases for field names to map fields from request to view arguments.

https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#alias-parameters

👇

from typing import Optional

from fastapi import FastAPI, Query, Path

app = FastAPI()


@app.get("/products/")
def search_products(query: Optional[str] = Query(None, alias="q")):
    products = [{"name": "Computer"}, {"name": "HDD"}]

    return {"results": [product for product in products if query in product["name"]]}


@app.get("/users/{id}/profile/")
def user_profile(user_id: int = Path(None, alias="id")):
    return {
        "id": user_id,
        "username": "johndoe"
    }

FastAPI Middleware


FastAPI tip:

You can add custom middleware to your app to do something before or after each request.

For example, to add a header containing the version of your application for easier debugging:

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_version_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Version"] = "v1.2.10"
    return response


@app.get("/ping")
def ping():
    return {"message": "pong"}

FastAPI - Using "callable" instances as dependencies in your API endpoints


FastAPI tip:

You can inject instances of a class as a dependency to your API endpoints, which you can then use when you as a configurable dependency.

You need to make instances callable via __call__.

https://fastapi.tiangolo.com/advanced/advanced-dependencies/#a-callable-instance

👇

from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel


class User(BaseModel):
    username: str
    groups: set[str]


users = [
    User(username='johndoe', groups={"admin"}),
    User(username='bobbuilder', groups={"builders"}),
]

app = FastAPI()


class AuthorizeUser:
    def __init__(self, allowed_groups: set[str]):
        self._allowed_groups = allowed_groups

    def __call__(self, username: str):
        try:
            user = next(user for user in users if user.username == username)
        except StopIteration:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

        if user.groups.isdisjoint(self._allowed_groups):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

        return user


@app.get("/only-admins")
def only_admins(user: User = Depends(AuthorizeUser(allowed_groups={"admin"}))):
    return {"message": f"User: {user.username} is in admin group."}


@app.get("/only-builders")
def only_admins(user: User = Depends(AuthorizeUser(allowed_groups={"builders"}))):
    return {"message": f"User: {user.username} is in builders group."}

FastAPI - API key authentication


FastAPI Tip:

You can protect API endpoints with an API key like so:

from fastapi import FastAPI, Body, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

api_keys = [
    "akljnv13bvi2vfo0b0bw"
]  # This is encrypted in the database

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")  # use token authentication


def api_key_auth(api_key: str = Depends(oauth2_scheme)):
    if api_key not in api_keys:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Forbidden"
        )


app = FastAPI()


@app.get("/protected", dependencies=[Depends(api_key_auth)])
def add_post() -> dict:
    return {
        "data": "You used a valid API key."
    }


####################################


# call API
import requests

url = "http://localhost:8000/protected"

# The client should pass the API key in the headers
headers = {
  'Content-Type': 'application/json',
  'Authorization': 'Bearer akljnv13bvi2vfo0b0bw'
}

response = requests.get(url, headers=headers)
print(response.text)  # => "You used a valid API key."

FastAPI - set cookie when returning a response


FastAPI tip:

You can set a cookie on the response by using .set_cookie(). Response must be added as a view argument.

https://fastapi.tiangolo.com/advanced/response-cookies/

👇

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/session/")
def cookie(response: Response):
    response.set_cookie(key="mysession", value="1242r")
    return {"message": "Wanna cookie?"}

FastAPI - Overriding dependencies while running tests


FastAPI tip:

You can override a dependency of your app while running tests with dependency_overrides.

https://fastapi.tiangolo.com/advanced/testing-dependencies/#use-the-appdependency_overrides-attribute

For example, to connect to a test database:

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
from sqlalchemy.orm import Session, sessionmaker, declarative_base
from sqlalchemy import create_engine, Column, Integer, String


SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)


Base.metadata.create_all(bind=engine)


class UserSchema(BaseModel):
    email: str

    class Config:
        orm_mode = True


app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=UserSchema)
def create_user(user_data: UserSchema, database_session: Session = Depends(get_db)):
    db_user = User(email=user_data.email)
    database_session.add(db_user)
    database_session.commit()

    return db_user


SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)


def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()


# THIS
app.dependency_overrides[get_db] = override_get_db
# THIS

client = TestClient(app)


def test_create_user():
    response = client.post(
        "/users/",
        json={"email": "[email protected]"},
    )
    assert response.status_code == 200

Asynchronous Background Tasks in FastAPI


FastAPI tip:

You can use FastAPI's BackGround Tasks to run simple tasks in the background.

👇

from fastapi import BackgroundTasks


def send_email(email, message):
    pass


@app.get("/")
async def ping(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, "[email protected]", "Hi!")
    return {"message": "pong!"}

Use Celery for CPU intensive tasks and when you need a task queue.

Filter by Topic

Show All

Django

Docker

FastAPI

Flask

Pytest

Python

TDD