10 промтов для FastAPI: эндпоинты, Pydantic и фоновая обработка как у профи

FastAPI — это не просто очередной веб-фреймворк для Python. Это экосистема, в которой асинхронность, автоматическая документация и строгая типизация сплетаются в единое целое. С момента выхода первой стабильной версии в 2019 году (автор — Себастьян Рамирес, ныне часть команды разработчиков Python Software Foundation) FastAPI стал стандартом де-факто для создания высоконагруженных API. По данным опроса JetBrains за 2025 год, более 28% Python-разработчиков используют FastAPI в коммерческих проектах — это почти в два раза больше, чем в 2022 году.

Но даже с таким мощным инструментом можно писать «так себе» код. Чтобы ваш API был не просто рабочим, а элегантным, производительным и легко поддерживаемым, нужно освоить несколько ключевых практик. В этой статье я собрал 10 промтов (запросов к нейросети) для генерации кода, которые помогут вам прокачать навыки работы с эндпоинтами, Pydantic и фоновыми задачами. Каждый промт — это конкретный сценарий из реальной жизни, с пояснением «почему так» и примером результата.

1. Базовый эндпоинт с валидацией через Pydantic

Задача: Создать эндпоинт POST для создания пользователя с валидацией email и возраста.

Промт:

Напиши на FastAPI эндпоинт /users/, который принимает JSON с полями name (строка), email (строка, валидация формата), age (целое, от 18 до 120). Используй Pydantic-модель. Верни созданный объект с добавленным полем id (UUID).

Результат:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, Field
import uuid

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=18, le=120)

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    age: int

users_db = {}

@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    user_id = str(uuid.uuid4())
    user_data = UserResponse(id=user_id, **user.dict())
    users_db[user_id] = user_data
    return user_data

Почему это важно: Pydantic автоматически генерирует ошибки валидации (например, при неверном email) и отдаёт их в формате JSON. Поле response_model автоматически фильтрует поля ответа — клиент не увидит пароль, если вы его случайно добавите.

2. Эндпоинт с query-параметрами и пагинацией

Задача: GET-эндпоинт для списка пользователей с пагинацией и фильтром.

Промт:

Создай эндпоинт /users/ с GET-методом. Поддержи page (int, default=1), limit (int, default=10, max=100) и search (опциональная строка). Верни список пользователей и общее количество.

Результат:

from typing import Optional
from fastapi import Query

@app.get("/users/")
async def list_users(
    page: int = Query(1, ge=1),
    limit: int = Query(10, ge=1, le=100),
    search: Optional[str] = None
):
    start = (page - 1) * limit
    end = start + limit

    if search:
        filtered = [u for u in users_db.values() if search.lower() in u.name.lower()]
    else:
        filtered = list(users_db.values())

    return {
        "total": len(filtered),
        "page": page,
        "limit": limit,
        "items": filtered[start:end]
    }

Совет: В реальном проекте используйте SQLAlchemy с LIMIT и OFFSET на уровне базы данных — это быстрее фильтрации в Python.

3. Фоновая задача с Celery

Задача: Эндпоинт, который запускает длительную обработку (например, генерация отчёта) в фоне.

Промт:

Напиши эндпоинт POST /reports/generate/, который принимает report_type и user_id. Запусти задачу Celery для генерации отчёта. Верни task_id. Укажи пример файла tasks.py.

Результат:

# main.py
from fastapi import BackgroundTasks, FastAPI
from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost:6379/0')

@app.post("/reports/generate/")
async def generate_report(report_type: str, user_id: int):
    task = celery_app.send_task('tasks.generate_report', args=[report_type, user_id])
    return {"task_id": task.id, "status": "queued"}

# tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def generate_report(report_type: str, user_id: int):
    import time
    time.sleep(10)  # симуляция длительной работы
    return f"Report {report_type} for user {user_id} generated"

Важно: Для production используйте Redis или RabbitMQ как брокер. Celery требует запущенного воркера: celery -A tasks worker --loglevel=info.

4. BackgroundTasks для простых фоновых задач

Задача: Отправить email после регистрации без Celery.

Промт:

Используй BackgroundTasks из FastAPI. Напиши эндпоинт POST /register/, который после создания пользователя запускает фоновую отправку email (симуляцию).

Результат:

from fastapi import BackgroundTasks

def send_welcome_email(email: str):
    import time
    time.sleep(2)
    print(f"Email sent to {email}")

@app.post("/register/")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
    user_id = str(uuid.uuid4())
    users_db[user_id] = user
    background_tasks.add_task(send_welcome_email, user.email)
    return {"id": user_id, "message": "User created. Email will be sent."}

Когда использовать: Для быстрых задач (отправка уведомлений, логирование), которые должны быть запущены после ответа. Для тяжёлых вычислений — Celery.

5. Валидация с кастомными полями Pydantic

Задача: Создать модель с кастомным валидатором для номера телефона.

Промт:

Добавь в Pydantic модель поле phone с валидацией: только цифры, 10-15 символов, начинается с +. Используй @validator.

Результат:

from pydantic import BaseModel, validator

class UserCreate(BaseModel):
    name: str
    phone: str

    @validator("phone")
    def validate_phone(cls, v):
        if not v.startswith("+"):
            raise ValueError("Phone must start with '+'")
        if not v[1:].isdigit():
            raise ValueError("Phone must contain only digits after '+'")
        if len(v) < 10 or len(v) > 15:
            raise ValueError("Phone length must be between 10 and 15 characters")
        return v

Нюанс: В Pydantic v2 синтаксис изменён — используйте @field_validator из pydantic.

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

Задача: Эндпоинт, который делает запрос к внешнему API и возвращает результат.

Промт:

Напиши асинхронный эндпоинт GET /weather/{city}. Используй httpx.AsyncClient для запроса к OpenWeatherMap. Обработай ошибки (таймаут, 404).

Результат:

import httpx
from fastapi import HTTPException

@app.get("/weather/{city}")
async def get_weather(city: str):
    api_key = "YOUR_API_KEY"
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, timeout=10.0)
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            raise HTTPException(status_code=504, detail="External API timeout")
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                raise HTTPException(status_code=404, detail="City not found")
            raise HTTPException(status_code=502, detail="External API error")

Замечание: Для production используйте пул соединений и кеш (например, Redis).

7. Dependency Injection для авторизации

Задача: Создать зависимость для проверки токена.

Промт:

Напиши зависимость get_current_user, которая извлекает токен из заголовка Authorization: Bearer <token>, декодирует JWT и возвращает user_id. Если токен невалиден — 401.

Результат:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, "SECRET_KEY", algorithms=["HS256"])
        return payload.get("sub")
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token"
        )

@app.get("/protected/")
async def protected_route(user_id: str = Depends(get_current_user)):
    return {"user_id": user_id}

Совет: Храните секреты в переменных окружения, а не в коде. Используйте python-jose для JWT.

8. Настройка CORS и middleware

Задача: Разрешить CORS для всех доменов в dev-режиме.

Промт:

Добавь CORS middleware в FastAPI приложение. Разреши все origins, методы и заголовки. Используй CORSMiddleware.

Результат:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Внимание: В production укажите конкретные origins, а не *. Иначе любой сайт сможет делать запросы к вашему API.

9. Обработка ошибок с кастомными исключениями

Задача: Создать кастомный exception handler для 404.

Промт:

Напиши кастомный обработчик для всех HTTPException. Возвращай JSON с полями error и details.

Результат:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": True,
            "detail": exc.detail
        }
    )

Фишка: Можно создать единый формат ошибок для всего API — это упрощает клиентскую обработку.

10. Тестирование с pytest и TestClient

Задача: Написать тест для эндпоинта создания пользователя.

Промт:

Используй TestClient из FastAPI. Напиши тест, который отправляет POST запрос с валидными данными и проверяет статус 201 и наличие поля id.

Результат:

from fastapi.testclient import TestClient

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users/",
        json={"name": "Test", "email": "test@example.com", "age": 25}
    )
    assert response.status_code == 201
    data = response.json()
    assert "id" in data
    assert data["name"] == "Test"

def test_create_user_invalid_age():
    response = client.post(
        "/users/",
        json={"name": "Test", "email": "test@example.com", "age": 17}
    )
    assert response.status_code == 422

Совет: Используйте pytest-asyncio для тестирования асинхронных эндпоинтов.

Заключение

Эти 10 промтов покрывают 80% типовых задач, с которыми сталкивается разработчик на FastAPI. Начните с базовой валидации и dependency injection, затем переходите к асинхронным запросам и фоновым задачам. Главное — не копируйте код слепо: адаптируйте его под свои бизнес-требования.

FastAPI — это фреймворк, который растёт вместе с вами. Сначала вы используете простые эндпоинты, потом добавляете Celery, потом — микросервисы на базе RabbitMQ. И на каждом этапе Pydantic и асинхронность остаются вашими верными спутниками.

Если хотите углубиться в тему — рекомендую официальную документацию FastAPI (fastapi.tiangolo.com) и книгу «FastAPI: Modern Python Web Development» от Bill Lubanovic. А для практики — попробуйте реализовать небольшой проект, например, API для хранения заметок с регистрацией и фоновой отправкой email. Это закрепит все навыки из этой статьи.

P.S. Если вы используете Telegram Bot API или другие внешние сервисы в своих проектах на FastAPI, помните, что ASI Biont поддерживает подключение к Telegram через API — подробнее на asibiont.com/courses. Это может существенно упростить интеграцию уведомлений.