FastAPI

To test the scripts provided in this page, you can use the following methods:

Using python venv:

python3 -m venv venv-fastapi
source venv-fastapi/bin/activate
pip install "fastapi[standard]" aiofiles

Save the code to test to the file simple_api.py and run the app using the command:

python3 -m fastapi dev simple_api.py
# OR
fastapi dev simple_api.py

Note

The default port used is 8000, you can change it using the argument --port (for example: fastapi dev simple_api.py --port 5000)

You can also use docker to run the APP:

1. Create the Dockerfile:

   FROM python:3.12-slim
   WORKDIR /app
   COPY simple_api.py /app
   RUN pip install "fastapi[standard]" aiofiles
   CMD ["python3", "-m", "fastapi", "dev", "simple_api.py"]
   

2. Save the code to test to the file simple_api.py

3. Run the Dockerfile:

   docker build -t fastapi-app . && docker run --rm --network host -it fastapi-app
   

Simple API

Receiving data from the client

Note

As explained in the official documentation , if you’re using Form to receive form data in the request, you need to make sure you installed the python package python-multipart (package automatically installed if you installed FastAPI using fastapi[standard])

Note

To save a file received in a form, we use the asynchronous way explained in this stackoverflow question , for that we need to install an additional python package aiofiles

from fastapi import FastAPI, Request, Header, Cookie, Form, UploadFile
from pydantic import BaseModel
from typing import Annotated
import aiofiles


app = FastAPI()

# Query parameters
@app.get("/query")
def query_paramters(p1: str, p2: str | None = None):
    print(f"p1: {p1}")
    print(f"p2 (optional): {p2}")
    return ""

# Path parameters
@app.post("/path/{username}/{post_id}")
def path_paramters(username: str, post_id: int):
    print(f'username: {username}, post_id: {post_id}')
    return ""

# Request body parameters
@app.post("/body_raw")
async def body_raw_parameters(request: Request):
    raw_data = await request.body()
    print(f'raw_data: {raw_data}')
    return ""

class Person(BaseModel):
    name: str
    age: float

@app.post("/body_json")
async def body_json_parameters(person: Person):
    print(person)
    return ""

# HTTP headers parameters
# List of headers: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers
@app.get("/header")
async def header_parameters(user_agent: Annotated[str | None, Header()] = None):
    print(user_agent)
    return ""

# Cookies parameters
@app.get("/cookies")
async def cookies_parameters(username: Annotated[str | None, Cookie()] = None):
    print(username)
    return ""

# Form Data (URL-encoded or Multipart)
@app.post("/form")
async def form_parameters(
    name: Annotated[str, Form()],
    email: Annotated[str, Form()],
    file_in: UploadFile | None = None,
    content_type: Annotated[str | None, Header()] = None):
    print(f"Content-Type: {content_type}")
    if 'multipart/form-data' in content_type:
        print(f"Content of 'file_in': {await file_in.read()}")
        await file_in.seek(0)
        async with aiofiles.open('uploaded_file.txt', 'wb') as out_file:
            content = await file_in.read()
            await out_file.write(content)
    print(f"name: {name}, email: {email}")
    return ""

You can test the routes using the curl commands below:

curl -X GET "http://localhost:8000/query?p1=value123"
curl -X POST "http://localhost:8000/path/nicoh/101"
curl -X POST "http://localhost:8000/body_raw" -d "This is raw data"
curl -X POST "http://localhost:8000/body_json" -H "Content-Type: application/json" -d '{"name": "Hania <3", "age": 30}'
curl -X GET "http://localhost:8000/header" -H "User-Agent: CustomUserAgent/1.0"
curl -X GET "http://localhost:8000/cookies" --cookie "username=nicoh"
curl -X POST "http://localhost:8000/form" -d "name=Nicolas&email=nicolas@example.com"
echo "Coucou" > file_test.txt
curl -X POST "http://localhost:8000/form" -F "name=Nicolas" -F "email=nicolas@example.com" -F "file_in=@file_test.txt"

Sending data to the client

from fastapi import FastAPI, status, Response, HTTPException
from pydantic import BaseModel
from fastapi.responses import HTMLResponse, PlainTextResponse, FileResponse, StreamingResponse
from typing import Any
from markupsafe import escape

app = FastAPI()

# HTML response
@app.get('/html/{name}', response_class=HTMLResponse)
async def html_response(name: str):
    # escape: To protect from injection attacks
    return f"<h1>Hello {escape(name)}</h1>"

# TEXT response
@app.get('/text', response_class=PlainTextResponse, status_code=status.HTTP_200_OK)
async def txt_response():
    return "This is the content"

# Explicit status code response
tasks = {"foo": "Listen to the Bar Fighters"}
@app.put("/get-or-create-task/{task_id}", status_code=200)
async def get_or_create_task(task_id: str, response: Response):
    if task_id not in tasks:
        if task_id == "error":
            raise HTTPException(status_code=404, detail="Item not found")
        tasks[task_id] = "This didn't exist before"
        response.status_code = status.HTTP_201_CREATED
    return tasks[task_id]

# json response
@app.get('/json_any')
async def json_response() -> Any:
    return {"message": "Hello, World!", "status": "success"}

class Person(BaseModel):
    name: str
    age: int

@app.get('/json_model')
async def json_response() -> Person:
    return {"name": "Hania <3", "age": 33}

# File
@app.get('/download')
async def download_file():
    filepath = "file_to_send.txt"
    with open(filepath, "w") as f:
        f.write("this is the file content")
    return FileResponse(filepath, filename="custom_filename.txt")

@app.get('/download_2')
async def download_file():
    filepath = "file_to_send.txt"
    with open(filepath, "w") as f:
        f.write("this is the file content 2")
    def iterfile():
        with open(filepath, mode="rb") as f:
            yield from f
    return StreamingResponse(iterfile(), media_type="text/plain")

You can test the routes using the curl commands below:

curl -X GET "http://localhost:8000/html/nicolas"
curl -X GET "http://localhost:8000/text"
curl -X PUT "http://localhost:8000/get-or-create-task/error"
curl -X PUT "http://localhost:8000/get-or-create-task/foo2"
curl -X PUT "http://localhost:8000/get-or-create-task/foo"
curl -X GET "http://localhost:8000/json_any"
curl -X GET "http://localhost:8000/json_model"
curl -X GET "http://localhost:8000/download"
curl -o downloaded_file.txt http://localhost:8000/download
curl -o downloaded_file_2.txt http://localhost:8000/download_2

Basic bearer token implementation

from fastapi import FastAPI, Depends, status
from starlette.requests import Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi.exceptions import HTTPException
from typing import Optional


app = FastAPI()


class BearerTokenChecker(HTTPBearer):
    def __init__(self, token: str):
        super().__init__()
        self.token = token

    async def __call__(self, request: Request) -> Optional[HTTPAuthorizationCredentials]:
        auth = await super().__call__(request)
        if auth is None:
            return None
        if auth.credentials != self.token:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Invalid authentication credentials",
            )
        return auth


security = BearerTokenChecker(token="super_secret_token_value")


@app.get('/secret', dependencies=[Depends(security)])
def get_secret():
    return "I love you"

You can test the route using the curl command below:

export API_AUTH_TOKEN=super_secret_token_value
curl -X GET -H "Authorization: Bearer $API_AUTH_TOKEN" "http://localhost:8000/secret"

Socket.IO

For the test below, I choose the most popular Socket.IO module for flask: fastapi-socketio (Github: https://github.com/pyropy/fastapi-socketio) You can install it using the command: pip install fastapi-socketio

from fastapi import FastAPI
from fastapi_socketio import SocketManager

app = FastAPI()
socket_manager = SocketManager(app=app, mount_location="/")


@app.sio.on('Custom Event')
async def test_message(sid, message):
    print(f"Message received: {message}")
    # By default, message are sent to the same client from which it received the message
    await app.sio.emit('Response FastAPI SocketIO', "Et voila ! Happy ?")
    # In 'emit' you can send a message to another client using argument 'to=other_client_sid'

@app.sio.on('my broadcast event')
async def test_message_broadcast(sid, message):
    # With broadcast=True, the message is sent to all connected clients
    await app.sio.emit('Response FastAPI SocketIO', {'data': message['data']}, broadcast=True)

@app.sio.on('connect')
async def handle_join(sid, *args, **kwargs):
    await app.sio.emit('Response FastAPI SocketIO', {'data': 'Connected'})
    print(f'Client with sid "{sid}" connected')

@app.sio.on('disconnect')
async def handle_leave(sid, *args, **kwargs):
    print(f'Client with sid "{sid}" disconnected')


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("simple_api:app", host='0.0.0.0', port=8000, reload=True)

You can test the routes using the python code below:

import socketio


sio = socketio.Client()

@sio.event
def connect():
    print('Connected to server')
    sio.emit('Custom Event', {'data': 'Hello from client'})

@sio.event
def disconnect():
    print('Disconnected from server')

@sio.on('Response Flask SocketIO')
def on_message(message):
    print(f"Received response: {message}")

sio.connect('http://localhost:5000')
sio.wait()

Rest API with OAuth & MFA

In this project, I provide an example of a REST FastAPI app that implements commonly used features following best practices.

Rest Flask Template: [WORK IN PROGRESS]

---

Sources:

• FastAPI documentation: https://fastapi.tiangolo.com

• fastapi-socketio documentation: https://github.com/pyropy/fastapi-socketio