Создание веб-сервисов с FastAPI: от основ до сложных решений

Пример 1: WebSocket соединение

FastAPI поддерживает WebSocket для двусторонней связи. Пример простого эхо-сервера:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f'Вы отправили: {data}')

Запуск сервера и подключение через JavaScript или wscat.

В консоли: wscat -c ws://localhost:8000/ws
> Привет
< Вы отправили: Привет

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

Иногда нужно выполнить действие после отправки ответа клиенту (например, отправка письма). FastAPI предоставляет встроенную возможность через BackgroundTasks.

from fastapi import BackgroundTasks

def write_log(message: str):
    with open('log.txt', 'a') as f:
        f.write(f'{message}\n')

@app.post('/send-notification')
async def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, 'Уведомление отправлено')
    return {'status': 'ok'}

Фоновая задача запускается после возврата ответа, не задерживая его.

Пример 3: Пагинация с использованием Limit/Offset

Для списков используется query-параметры limit и offset. Расширенный пример с инъекцией зависимостей:

from fastapi import Query, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def paginate(db: AsyncSession, model, skip: int = 0, limit: int = 10):
    query = select(model).offset(skip).limit(limit)
    result = await db.execute(query)
    return result.scalars().all()

@app.get('/items')
async def list_items(skip: int = Query(0, ge=0), limit: int = Query(10, le=100), db = Depends(get_db)):
    items = await paginate(db, Item, skip, limit)
    return items

Пример 4: Загрузка файлов с валидацией

FastAPI использует UploadFile для работы с загружаемыми файлами.

from fastapi import File, UploadFile

@app.post('/upload/')
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # обработка содержимого
    return {'filename': file.filename, 'size': len(contents)}

Можно добавить проверку типа файла через file.content_type.

Пример 5: Настройка CORS для доступа с другого домена

Использование CORSMiddleware из fastapi.middleware.cors.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=['https://frontend.example.com'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

Пример 6: Middleware для логирования времени ответа

Собственный middleware позволяет перехватывать запросы и ответы.

from fastapi import Request
import time

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers['X-Process-Time'] = str(process_time)
    return response

Пример 7: Тестирование с pytest и httpx

Для тестирования используется TestClient из Starlette (встроен в FastAPI).

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get('/')
    assert response.status_code == 200
    assert response.json() == {'message': 'Hello World'}

Для асинхронных тестов нужно использовать pytest-asyncio.

Пример 8: Docker-образ для FastAPI приложения

Пример Dockerfile:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Сборка и запуск: docker build -t myapi . && docker run -p 8000:8000 myapi.