Практические задачи FastAPI.md

Начальный уровень

1. Создание простого эндпоинта

Задание: Напишите эндпоинт GET /hello, который возвращает {"message": "Hello, World!"}.
Проверяемые знания: Базовое создание роута, работа с декораторами, возврат JSON.

2. Параметры пути и запроса

Задание: Создайте эндпоинт GET /items/{item_id}, который принимает item_id (путь) и q (query-параметр, опциональный). Возвращает JSON с этими значениями.
Проверяемые знания: Path-параметры, query-параметры, валидация.

3. POST-запрос с валидацией данных

Задание: Сделайте эндпоинт POST /items/, который принимает JSON с полями name (обязательный, строка) и price (обязательный, положительное число). Используйте Pydantic для валидации.
Проверяемые знания: Pydantic модели, валидация, POST-запросы.

4. Обработка ошибок

Задание: Создайте кастомный обработчик ошибки для HTTP 404, который возвращает JSON {"error": "Not Found", "detail": "Item not found"}.
Проверяемые знания: Кастомные исключения, обработчики ошибок.

5. Зависимости (Dependencies)

Задание: Реализуйте dependency get_db, который "подключается" к БД (можно имитировать) и возвращает соединение. Используйте его в эндпоинте GET /users/.
Проверяемые знания: DI (Dependency Injection), работа с зависимостями.

6. Асинхронный эндпоинт

Задание: Напишите асинхронный эндпоинт GET /slow/, который ждет 1 секунду (через asyncio.sleep), затем возвращает {"status": "ok"}.
Проверяемые знания: Асинхронность, async/await.

7. Файлы и загрузка данных

Задание: Сделайте эндпоинт POST /upload/, который принимает файл и возвращает его размер в байтах.
Проверяемые знания: Работа с файлами (UploadFile), FormData.

8. Middleware

Задание: Добавьте middleware, который добавляет заголовок X-Process-Time с временем обработки запроса в миллисекундах.
Проверяемые знания: Middleware, время выполнения запроса.

9. JWT-аутентификация

Задание: Реализуйте эндпоинт /login, который принимает username/password и возвращает JWT-токен. Затем сделайте защищенный эндпоинт /me, требующий этот токен.
Проверяемые знания: JWT, OAuth2, безопасность.

10. Тестирование эндпоинтов

Задание: Напишите тест (с помощью pytest) для эндпоинта GET /hello, проверяющий статус-код и ответ.
Проверяемые знания: Тестирование FastAPI (TestClient), pytest.

Бонус (для продвинутых):

Фоновые задачи: Создайте эндпоинт, который запускает фоновую задачу (например, пишет в лог через 5 секунд).
WebSockets: Реализуйте простой WebSocket-эндпоинт (/ws), который возвращает клиенту его же сообщения.

Продвинутый уровень

11. Кастомные валидаторы в Pydantic

Задание: Создайте модель UserCreate, где:
- Пароль (password) должен содержать минимум 8 символов, включая цифру и спецсимвол.
- Email проверяется через регулярное выражение.
Цель: Реализация кастомных валидаторов с использованием @validator в Pydantic.

12. Оптимизация запросов к БД с помощью Dependency

Задание:
1. Создайте dependency get_db, который открывает транзакцию.
2. Реализуйте эндпоинт GET /users/{user_id}/posts, который возвращает пользователя и его посты в одном запросе к БД (без N+1 проблемы).
Проверяемые знания: SQLAlchemy (или другой ORM), eager loading, транзакции.

13. Rate Limiting (лимит запросов)

Задание: Настройте лимит в 5 запросов в минуту для эндпоинта POST /comments/ с помощью:
- Redis для хранения счетчика.
- Кастомного middleware или dependency.
Цель: Работа с внешними сервисами, асинхронные вызовы.

14. Генерация OpenAPI схемы программно

Задание: Динамически добавьте в OpenAPI схему кастомный параметр X-API-Version для всех эндпоинтов.
Проверяемые знания: Кастомизация OpenAPI, доступ к app.openapi().

15. Асинхронная задача в Celery

Задание:
1. Создайте эндпоинт POST /report/, который запускает фоновую задачу в Celery на генерацию отчета.
2. Эндпоинт GET /report/{task_id} должен возвращать статус задачи.
Цель: Интеграция FastAPI с Celery, работа с AsyncResult.

16. OAuth2 с кастомной логикой

Задание: Реализуйте OAuth2 авторизацию, где:
- Токен выдается только пользователям с ролью admin.
- В токене должен быть user_id и scopes.
Проверяемые знания: OAuth2PasswordBearer, JWT, кастомные зависимости.

17. Тестирование с моками БД и внешних API

Задание: Напишите тест для эндпоинта GET /weather/{city}, который:
- Мокает запрос к внешнему API (например, OpenWeatherMap).
- Проверяет обработку ошибки 404.
Цель: pytest, pytest-mock, httpx.AsyncClient.

18. WebSocket с комнатами чата

Задание: Создайте WebSocket (/chat/{room_id}), где:
- Сообщения от одного пользователя рассылаются всем в комнате.
- История сообщений хранится в Redis.
Проверяемые знания: WebSocket, бродкастинг, работа с Redis.

19. GraphQL в FastAPI (Strawberry)

Задание: Интегрируйте GraphQL через Strawberry:
- Создайте тип Book с полями id, title, author.
- Реализуйте query getBooks и mutation addBook.
Цель: GraphQL, интеграция сторонних библиотек.

20. Оптимизация ответов с помощью ETag

Задание: Для эндпоинта GET /products/ добавьте заголовок ETag на основе хеша данных.
- Если ETag совпадает с If-None-Match, верните 304 Not Modified.
Проверяемые знания: HTTP-кеширование, кастомные заголовки.

Бонус (экспертный уровень):

gRPC + FastAPI: Создайте gRPC сервис для проверки орфографии, который вызывается из FastAPI.
Server-Sent Events (SSE): Реализуйте эндпоинт /stream, который отправляет события каждые 5 секунд.

Экспертный уровень

21. Оптимизация GraphQL-запросов (DataLoader & N+1)

Задача:

Реализуйте GraphQL API (через Strawberry или Ariadne) с типами User и Post.
Используйте DataLoader для решения проблемы N+1 при запросе списка пользователей с их постами.
Что проверяем:
Понимание проблем N+1 в GraphQL.
Умение работать с асинхронными DataLoader.

22. Кастомная валидация запросов с помощью Middleware

Задача:

Создайте middleware, который:
- Блокирует запросы с подозрительными User-Agent (например, curl или Postman).
- Добавляет кастомный заголовок X-Request-Source.
  Что проверяем:
Глубокое знание middleware и работы с заголовками.
Безопасность и защиту от скрейпинга.

23. Динамическое подключение/отключение эндпоинтов

Задача:

Реализуйте административный эндпоинт POST /admin/toggle_route, который:
- Динамически включает/выключает другие эндпоинты (например, /api/status).
  Что проверяем:
Работу с app.routes, динамическое изменение API.
Паттерн Feature Toggle.

24. Асинхронная работа с Kafka

Задача:

Создайте эндпоинт POST /events/, который отправляет событие в Kafka.
Реализуйте consumer в фоне (через asyncio), который обрабатывает эти события и пишет их в БД.
Что проверяем:
Интеграцию с брокерами сообщений (Kafka/RabbitMQ).
Асинхронную обработку событий.

25. Мультитенантность (Multi-Tenancy) в FastAPI

Задача:

Реализуйте систему, где каждый клиент (tenant) использует разные БД.
Автоматически определяйте tenant из заголовка X-Tenant-ID и подключайте соответствующую БД.
Что проверяем:
Dependency Injection для мультитенантности.
Динамическое подключение к разным БД (SQLAlchemy/asyncpg).

26. Автоматическая документация с кастомными тегами

Задача:

Генерируйте OpenAPI-схему так, чтобы:
- Эндпоинты группировались по кастомным тегам (например, Legacy и V2).
- Добавлялось описание для каждого тега.
  Что проверяем:
Кастомизацию OpenAPI, работу с app.openapi_schema.

27. Распределённые транзакции (Saga Pattern)

Задача:

Реализуйте эндпоинт POST /orders/, который:
1. Создает заказ в БД.
2. Резервирует товар на складе (через HTTP-вызов).
3. Если склад недоступен — откатывает заказ.
  Что проверяем:
Паттерн Saga для распределённых систем.
Обработку ошибок и компенсирующие транзакции.

28. Server-Sent Events (SSE) + Redis Pub/Sub

Задача:

Создайте эндпоинт GET /notifications/, который:
- Подключается к Redis Pub/Sub.
- Отправляет клиенту уведомления в реальном времени через SSE.
  Что проверяем:
Работу с Server-Sent Events.
Интеграцию с Redis для real-time уведомлений.

29. gRPC-микросервис + FastAPI Gateway

Задача:

Создайте gRPC-сервис для проверки текста на спам.
Интегрируйте его в FastAPI через grpclib.
Добавьте кеширование результатов в Redis.
Что проверяем:
Работу с gRPC и protobuf.
Паттерн API Gateway.

30. Оптимизация через Cython или Rust

Задача:

Вынесите CPU-bound логику (например, хеширование паролей) в Rust (через pyo3) или Cython.
Интегрируйте модуль в FastAPI.
Что проверяем:
Умение оптимизировать критические участки кода.
Работу с FFI (Foreign Function Interface).

Бонус (для супер-экспертов):

Реализация JIT-компиляции эндпоинтов: Динамически генерируйте эндпоинты на основе конфигурации в runtime.
Своя система плагинов: FastAPI-приложение, которое подгружает эндпоинты из внешних Python-модулей.

Найти ошибку в коде - Начальный

Задача 1: Простой эндпоинт

Код

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id):
    return {"item_id": item_id}

Задача 2: Параметры запроса

Код

from fastapi import FastAPI

app = FastAPI()

@app.get("/search")
def search(query):
    return {"query": query}

Задача 3: Валидация данных с Pydantic

Код

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
def create_item(item: Item):
    return item

Задача 4: Асинхронный эндпоинт

Код

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async")
async def async_task():
    await asyncio.sleep(2)
    return {"message": "Task completed"}

Задача 5: Загрузка файлов

Код

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload")
def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

Задача 6: Background Tasks

Код

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

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

@app.post("/send-email")
def send_email(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Email sent")
    return {"message": "Email sent in the background"}

Задача 7: Middleware

Код

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Process-Time"] = "10ms"
    return response

Задача 8: WebSocket

Код

from fastapi import FastAPI, WebSocket

app = FastAPI()

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

Задача 9: CORS

Код

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET"],
)

Задача 10: Пагинация

Код

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
def get_items(page: int = Query(1), limit: int = Query(10)):
    items = [{"id": i} for i in range((page - 1) * limit, page * limit)]
    return {"items": items}

Найти ошибку в коде - Продвинутый

Задача 11: Асинхронная работа с базой данных

Код

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()

DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    db = SessionLocal()
    user = db.execute("SELECT * FROM users WHERE id = :id", {"id": user_id}).fetchone()
    return user

Задача 12: Асинхронные фоновые задачи

Код

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def long_task():
    time.sleep(10)

@app.post("/task")
def run_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(long_task)
    return {"message": "Task started"}

Задача 13: Потоковая передача данных

Код

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

@app.get("/stream")
def stream_data():
    def generate():
        for i in range(10):
            time.sleep(1)
            yield f"data: {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Задача 14: Работа с Redis

Код

from fastapi import FastAPI
import redis

app = FastAPI()

redis_client = redis.Redis(host="localhost", port=6379, db=0)

@app.get("/cache/{key}")
def get_cache(key: str):
    value = redis_client.get(key)
    return {"key": key, "value": value}

Задача 15: WebSocket и Pub/Sub

Код

from fastapi import FastAPI, WebSocket
import redis

app = FastAPI()

redis_client = redis.Redis(host="localhost", port=6379, db=0)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    pubsub = redis_client.pubsub()
    pubsub.subscribe("channel")
    for message in pubsub.listen():
        await websocket.send_text(message["data"])

Задача 16: Аутентификация через JWT

Код

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "mysecret"

@app.get("/protected")
def protected_route(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {"user_id": payload["user_id"]}
    except:
        raise HTTPException(status_code=401, detail="Invalid token")

Задача 17: GraphQL API

Код

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello, world!"

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Задача 18: Масштабируемость приложения

Код

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def get_items():
    # Логика получения данных
    return {"items": []}

Задача 19: Интеграция с внешним API

Код

from fastapi import FastAPI
import requests

app = FastAPI()

@app.get("/weather")
def get_weather(city: str):
    response = requests.get(f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=your_api_key")
    return response.json()

Задача 20: Event-Driven Architecture

Код

from fastapi import FastAPI
from kafka import KafkaProducer

app = FastAPI()

producer = KafkaProducer(bootstrap_servers="localhost:9092")

@app.post("/events")
def send_event(event: dict):
    producer.send("topic", value=event)
    return {"message": "Event sent"}

Найти ошибку в коде - Экспертный

Задача 21: Асинхронная обработка больших файлов

Код

from fastapi import FastAPI, UploadFile
import pandas as pd

app = FastAPI()

@app.post("/process-file")
async def process_file(file: UploadFile):
    df = pd.read_csv(file.file)
    result = df.groupby("category").sum()
    return {"summary": result.to_dict()}

Задача 22: Масштабируемая система авторизации

Код

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "mysecret"

@app.get("/protected")
def protected_route(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {"user_id": payload["user_id"]}
    except:
        raise HTTPException(status_code=401, detail="Invalid token")

Задача 23: Event-Driven Architecture с Kafka

Код

from fastapi import FastAPI
from kafka import KafkaProducer

app = FastAPI()

producer = KafkaProducer(bootstrap_servers="localhost:9092")

@app.post("/events")
def send_event(event: dict):
    producer.send("topic", value=event)
    return {"message": "Event sent"}

Задача 24: Работа с WebSocket и Redis Streams

Код

from fastapi import FastAPI, WebSocket
import redis

app = FastAPI()

redis_client = redis.Redis(host="localhost", port=6379, db=0)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    pubsub = redis_client.pubsub()
    pubsub.subscribe("channel")
    for message in pubsub.listen():
        await websocket.send_text(message["data"])

Задача 25: Высокопроизводительное стриминговое API

Код

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

@app.get("/stream")
async def stream_data():
    async def generate():
        for i in range(10):
            await asyncio.sleep(1)
            yield f"data: {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Задача 26: Реализация CQRS

Код

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()

DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    db = SessionLocal()
    user = db.execute("SELECT * FROM users WHERE id = :id", {"id": user_id}).fetchone()
    return user

@app.post("/users")
def create_user(name: str, email: str):
    db = SessionLocal()
    db.execute("INSERT INTO users (name, email) VALUES (:name, :email)", {"name": name, "email": email})
    db.commit()
    return {"message": "User created"}

Задача 27: Развертывание в Kubernetes

Код

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastapi
  template:
    metadata:
      labels:
        app: fastapi
    spec:
      containers:
      - name: fastapi
        image: my-fastapi-app:latest
        ports:
        - containerPort: 8000

Задача 28: GraphQL API с подписками

Код

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello, world!"

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Задача 29: Микросервисная архитектура

Код

from fastapi import FastAPI
from httpx import AsyncClient

app = FastAPI()

@app.get("/users/{user_id}/orders")
async def get_user_orders(user_id: int):
    async with AsyncClient() as client:
        response = await client.get(f"http://order-service/orders?user_id={user_id}")
        return response.json()

Задача 30: Rate Limiting

Код

from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/limited")
@limiter.limit("5/minute")
def limited_route():
    return {"message": "This route is rate-limited"}

Eugene-Fed/Практические задачи FastAPI.md

Оглавление

Начальный уровень

1. Создание простого эндпоинта

2. Параметры пути и запроса

3. POST-запрос с валидацией данных

4. Обработка ошибок

5. Зависимости (Dependencies)

6. Асинхронный эндпоинт

7. Файлы и загрузка данных

8. Middleware

9. JWT-аутентификация

10. Тестирование эндпоинтов

Бонус (для продвинутых):

Продвинутый уровень

11. Кастомные валидаторы в Pydantic

12. Оптимизация запросов к БД с помощью Dependency

13. Rate Limiting (лимит запросов)

14. Генерация OpenAPI схемы программно

15. Асинхронная задача в Celery

16. OAuth2 с кастомной логикой

17. Тестирование с моками БД и внешних API

18. WebSocket с комнатами чата

19. GraphQL в FastAPI (Strawberry)

20. Оптимизация ответов с помощью ETag

Бонус (экспертный уровень):

Экспертный уровень

21. Оптимизация GraphQL-запросов (DataLoader & N+1)

22. Кастомная валидация запросов с помощью Middleware

23. Динамическое подключение/отключение эндпоинтов

24. Асинхронная работа с Kafka

25. Мультитенантность (Multi-Tenancy) в FastAPI

26. Автоматическая документация с кастомными тегами

27. Распределённые транзакции (Saga Pattern)

28. Server-Sent Events (SSE) + Redis Pub/Sub

29. gRPC-микросервис + FastAPI Gateway

30. Оптимизация через Cython или Rust

Бонус (для супер-экспертов):

Найти ошибку в коде - Начальный

Задача 1: Простой эндпоинт

Код

Задача 2: Параметры запроса

Код

Задача 3: Валидация данных с Pydantic

Код

Задача 4: Асинхронный эндпоинт

Код

Задача 5: Загрузка файлов

Код

Задача 6: Background Tasks

Код

Задача 7: Middleware

Код

Задача 8: WebSocket

Код

Задача 9: CORS

Код

Задача 10: Пагинация

Код

Найти ошибку в коде - Продвинутый

Задача 11: Асинхронная работа с базой данных

Код

Задача 12: Асинхронные фоновые задачи

Код

Задача 13: Потоковая передача данных

Код

Задача 14: Работа с Redis

Код

Задача 15: WebSocket и Pub/Sub

Код

Задача 16: Аутентификация через JWT

Код

Задача 17: GraphQL API

Код

Задача 18: Масштабируемость приложения

Код

Задача 19: Интеграция с внешним API

Код

Задача 20: Event-Driven Architecture

Код