Создание быстрых и надёжных API с помощью FastAPI

Реализация WebSocket соединения

FastAPI поддерживает WebSocket. Пример эхо-сервера:

from fastapi import WebSocket

@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}")

Клиент подключается через ws://localhost:8000/ws и отправляет сообщения. В ответ приходит то же сообщение с префиксом.

(Browser console)
> ws = new WebSocket("ws://localhost:8000/ws")
> ws.send("Привет")
> ws.onmessage = (e) => console.log(e.data)
< "Message text was: Привет"

Фоновая задача с использованием BackgroundTasks

Для выполнения операций после отправки ответа (например, отправка email) используется BackgroundTasks:

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@app.post("/send-notification")
async def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "notification sent")
    return {"message": "Notification will be sent"}

Функция write_log выполняется в фоне после того, как ответ уже отправлен клиенту.

Загрузка файлов через multipart/form-data

FastAPI поддерживает загрузку файлов с помощью UploadFile:

from fastapi import File, UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # сохраняем файл
    with open(f"{file.filename}", "wb") as f:
        f.write(contents)
    return {"filename": file.filename, "size": len(contents)}

Клиент отправляет файл через форму. Размер файла можно получить до полного чтения, используя file.file.seek(0, 2) и file.file.tell().

POST /uploadfile/
Content-Type: multipart/form-data
...
Response: {"filename": "photo.jpg", "size": 123456}

Пагинация через параметры запроса

Реализация пагинации с помощью Query и Pydantic:

from pydantic import BaseModel
from fastapi import Query

class Pagination(BaseModel):
    page: int = Query(1, ge=1)
    per_page: int = Query(10, ge=1, le=100)

@app.get("/items")
def list_items(pagination: Pagination = Depends()):
    # имитация базы данных
    items = list(range(1, 1001))
    start = (pagination.page - 1) * pagination.per_page
    end = start + pagination.per_page
    return {"page": pagination.page, "items": items[start:end]}

GET /items?page=2&per_page=5
Response: {"page": 2, "items": [6, 7, 8, 9, 10]}

Кэширование ответов с использованием Redis

Пример простого декоратора для кэширования:

import json
import aioredis
from functools import wraps

redis = aioredis.from_url("redis://localhost")

def cache(expire: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{json.dumps(kwargs)}"
            cached = await redis.get(key)
            if cached:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await redis.set(key, json.dumps(result), ex=expire)
            return result
        return wrapper
    return decorator

@app.get("/cached")
@cache(expire=30)
async def get_cached_data():
    # имитация дорогого запроса
    await asyncio.sleep(2)
    return {"data": "very expensive"}

Первый запрос выполняется 2 секунды, последующие (в течение 30 секунд) возвращают закэшированный ответ мгновенно.

First request: 2 sec delay
Second request (within 30 sec): instant response