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

Современная разработка API на Python немыслима без FastAPI. Этот фреймворк сочетает скорость Starlette, мощную валидацию Pydantic и автоматическую генерацию OpenAPI-документации, что делает его выбором №1 для создания высоконагруженных сервисов. Однако даже опытные разработчики часто тратят часы на рутинные задачи: настройка эндпоинтов, модели данных, фоновые процессы. В этой статье я собрал 12 практичных промтов — готовых формулировок запросов к ИИ-ассистенту, которые помогут вам быстро решить типовые задачи FastAPI. Каждый промт сопровождается кодом, пояснением и ссылкой на официальную документацию. Вы узнаете, как правильно проектировать эндпоинты, использовать Pydantic для валидации и организовывать фоновую обработку с помощью async-возможностей FastAPI.

Базовые промты: создание эндпоинтов

1. Простой GET-эндпоинт для проверки здоровья

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

Промт: «Напиши FastAPI-приложение с эндпоинтом GET /health, который возвращает JSON-объект {'status': 'ok'}».

Пример результата:

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Это базовый шаблон, который можно расширить проверкой подключения к базе данных или внешним сервисам. В реальном проекте такой эндпоинт используется для балансировщиков нагрузки и оркестраторов (Kubernetes, Docker Compose).

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

Задача: Создать эндпоинт поиска, который принимает параметры через URL.

Промт: «Создай эндпоинт GET /items, принимающий query-параметры: q (строка поиска), limit (int, дефолт 10), offset (int, дефолт 0). Возвращает список items и общее количество».

Пример результата:

from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

items = [{"id": i, "name": f"Item {i}"} for i in range(100)]

@app.get("/items")
async def search_items(
    q: Optional[str] = Query(None, description="Поисковый запрос"),
    limit: int = Query(10, ge=1, le=100),
    offset: int = Query(0, ge=0)
):
    filtered = [item for item in items if q is None or q.lower() in item["name"].lower()]
    return {"total": len(filtered), "items": filtered[offset:offset+limit]}

Здесь используется Query для валидации параметров: ge=1 и le=100 ограничивают limit, что предотвращает некорректные запросы. FastAPI автоматически отобразит эти параметры в Swagger UI.

3. POST-эндпоинт с телом запроса

Задача: Создать эндпоинт для создания нового ресурса с валидацией данных.

Промт: «Создай модель Pydantic ItemCreate с полями name (str), price (float), is_offer (bool = False). Напиши эндпоинт POST /items, который принимает ItemCreate и возвращает созданный объект с id».

Пример результата:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ItemCreate(BaseModel):
    name: str
    price: float
    is_offer: bool = False

items_db = []

@app.post("/items")
async def create_item(item: ItemCreate):
    item_dict = item.dict()
    item_dict["id"] = len(items_db) + 1
    items_db.append(item_dict)
    return item_dict

Pydantic автоматически проверяет типы, преобразует входные данные и возвращает понятные ошибки клиенту. Например, если передать price="abc", клиент получит HTTP 422 с детализацией ошибки.

Продвинутые промты: валидация Pydantic

4. Кастомный валидатор для email

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

Промт: «Добавь в модель Pydantic UserCreate поле email, которое валидируется с помощью EmailStr. Обработай ошибку: если email некорректный, вернуть сообщение „Invalid email address“».

Пример результата:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, ValidationError

app = FastAPI()

class UserCreate(BaseModel):
    email: EmailStr
    password: str

@app.post("/register")
async def register(user: UserCreate):
    # Здесь можно добавить проверку уникальности email
    return {"message": "User created", "email": user.email}

EmailStr использует библиотеку email-validator, которая проверяет не только формат, но и существование домена. Для работы нужно установить pydantic[email] (подробнее в документации Pydantic).

5. Response model для фильтрации полей

Задача: Скрыть чувствительные данные (например, пароль) при возврате объекта из API.

Промт: «Определи модель UserInDB (содержит пароль) и UserOut (без пароля). Используй response_model=UserOut в эндпоинте GET /users/{user_id}».

Пример результата:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserInDB(BaseModel):
    id: int
    username: str
    password: str

class UserOut(BaseModel):
    id: int
    username: str

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    user = {"id": user_id, "username": "johndoe", "password": "secret"}
    return user  # Вернётся только id и username

response_model гарантирует, что даже если вы случайно вернёте лишние поля, они будут отфильтрованы. Это критически важно для защиты данных.

6. Зависимости (Depends) для аутентификации

Задача: Создать общий механизм проверки API-ключа.

Промт: «Напиши функцию-зависимость verify_token, которая читает заголовок X-Token и возвращает user_id. Примени её к нескольким эндпоинтам через Depends».

Пример результата:

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

def verify_token(x_token: str = Header(default=None)):
    if x_token != "secret123":
        raise HTTPException(status_code=401, detail="Invalid token")
    return 42  # user_id

@app.get("/profile")
async def get_profile(user_id: int = Depends(verify_token)):
    return {"user_id": user_id}

@app.get("/settings")
async def get_settings(user_id: int = Depends(verify_token)):
    return {"theme": "dark"}

Система зависимостей FastAPI позволяет переиспользовать логику аутентификации, ограничения доступа и другие кросс-функциональные сценарии. Это описано в официальном руководстве.

Экспертные промты: фоновая обработка и async

7. Фоновая задача без блокировки ответа

Задача: После регистрации пользователя отправить письмо, не заставляя клиента ждать.

Промт: «Используй BackgroundTasks в FastAPI, чтобы после POST /register выполнялась функция send_welcome_email, которая имитирует отправку письма (time.sleep). Ответ клиенту должен возвращаться немедленно».

Пример результата:

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def send_welcome_email(email: str):
    time.sleep(5)  # Эмуляция SMTP-запроса
    print(f"Welcome email sent to {email}")

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"message": "Registration successful"}

BackgroundTasks идеально подходит для задач, не требующих результата: отправка email, обновление кэша, запись в аналитику. Важно: задачи выполняются после отправки ответа, поэтому клиент не ждёт. Подробнее — в документации.

8. Async-эндпоинт с внешним HTTP-запросом

Задача: Получить данные от внешнего API, не блокируя event loop.

Промт: «Напиши async-эндпоинт GET /news, который с помощью aiohttp запрашивает данные с newsapi.example.com и возвращает заголовки новостей».

Пример результата:

from fastapi import FastAPI
import aiohttp

app = FastAPI()

NEWS_API_URL = "https://newsapi.example.com/top-headlines"

@app.get("/news")
async def get_news():
    async with aiohttp.ClientSession() as session:
        async with session.get(NEWS_API_URL) as resp:
            data = await resp.json()
            return {"headlines": [article["title"] for article in data["articles"]]}

Благодаря async/await FastAPI может обрабатывать сотни параллельных запросов, не создавая поток. Для работы с внешними API часто используют библиотеку httpx, которая поддерживает и async, и sync режимы (см. инструкцию).

9. Интеграция с async SQLAlchemy

Задача: Подключить базу данных PostgreSQL и выполнять асинхронные запросы.

Промт: «Создай async-сессию SQLAlchemy, модель Book, эндпоинт GET /books, который возвращает список книг из БД».

Пример результата:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

class Book(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author = Column(String)

app = FastAPI()

@app.get("/books")
async def get_books():
    async with SessionLocal() as session:
        result = await session.execute(select(Book))
        books = result.scalars().all()
        return [{"id": b.id, "title": b.title, "author": b.author} for b in books]

Такой подход позволяет не блокировать потоки при обращении к БД. Возможно также использовать asyncpg напрямую, но SQLAlchemy даёт ORM-абстракцию и миграции (Alembic).

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

Задача: Вернуть понятное сообщение об ошибке, когда ресурс не найден.

Промт: «Создай кастомное исключение NotFoundError и обработчик, который возвращает JSON с полем detail и кодом 404».

Пример результата:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class NotFoundError(Exception):
    pass

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return JSONResponse(status_code=404, content={"detail": "Resource not found"})

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id > 100:
        raise NotFoundError()
    return {"item_id": item_id}

Централизованная обработка исключений упрощает поддержку API. В FastAPI также есть встроенный класс HTTPException, но кастомные обработчики позволяют логировать ошибки или менять формат ответа, не затрагивая бизнес-логику.

11. Middleware для логирования запросов

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

Промт: «Добавь middleware, который логирует метод, URL и статус-код ответа с помощью time.perf_counter».

Пример результата:

from fastapi import FastAPI, Request
import time
import logging

app = FastAPI()
logging.basicConfig(level=logging.INFO)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    logging.info(f"{request.method} {request.url.path} -> {response.status_code} ({duration:.3f}s)")
    return response

Middleware полезен для метрик, трейсинга и аудита. В FastAPI можно также подключать стандартные ASGI-промежуточные слои через @app.middleware("http").

12. Веб-сокеты для real-time приложений

Задача: Реализовать чат-комнату на WebSocket.

Промт: «Напиши WebSocket-эндпоинт /ws, который принимает имя клиента, сохраняет соединение и транслирует сообщения всем подключённым клиентам».

Пример результата:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("Client left the chat")

FastAPI полностью поддерживает WebSocket-протокол, что делает его удобным для чатов, уведомлений и онлайн-игр. Пример из документации доступен в официальном руководстве.

Заключение

Мы разобрали 12 промтов, которые покрывают ключевые аспекты разработки на FastAPI: от простых эндпоинтов до сложных асинхронных сценариев. Эти шаблоны — стартовая точка для ваших проектов, их легко адаптировать под конкретные задачи. Помните, что FastAPI — это не только удобный синтаксис, но и мощная экосистема: Pydantic для валидации, SQLAlchemy для БД, BackgroundTasks для фоновой обработки. Если вы только начинаете, рекомендую изучить официальную документацию FastAPI — она считается одной из лучших в мире Python-разработки.

А теперь — практикуйтесь! Возьмите один из промтов, модифицируйте его и добавьте в свой проект. Если вы хотите систематизировать знания, обратите внимание на наш блог asibiont.com — там мы публикуем материалы по FastAPI, Python и современной веб-разработке. Подписывайтесь, чтобы не пропустить новые разборы и практические руководства.

← Все статьи

Комментарии

Читайте также

ESP32-S3 + ASI Biont: интеграция on-device ML с AI-агентом для умного дома

31 июля 2026

Интеграция ИИ с Тинькофф Банком: автоматизация счетов, платежей и денежных потоков с ASI Biont

31 июля 2026

Zigbee2MQTT + ASI Biont: как подключить Zigbee-устройства к AI-агенту и автоматизировать умный дом

31 июля 2026

RSS-to-email digests + ASI Biont: автоматизация дайджестов новостей через ИИ-агента без кода

31 июля 2026

Что происходит с интернетом, когда роботы ведут себя как люди: эпоха ИИ-агентов

31 июля 2026

Обзор курса Cloud Native: микросервисы, Kubernetes и облачные технологии от Asibiont

31 июля 2026

Готовое решение из коробки: финал саги — мой ИИ-ассистент на Hermes улетел на GitHub. Забирайте! (а грабли расскажу следом)

31 июля 2026

Gemini Robotics ER 2: видео-понимание, оркестрация задач и мульти-роботная коллаборация — как это работает

31 июля 2026

Подключение ЧПУ-станка к AI-агенту: GRBL + Marlin + ASI Biont — автоматизация производства через чат

31 июля 2026