Пример полной валидации данных с помощью Pydantic

На предыдущих лекциях мы освоили основы работы с Pydantic. Мы узнали, что это мощный инструмент для валидации данных в FastAPI, который превращает сырые данные в структурированные объекты, проверяя их корректность. Мы создали базовые Pydantic-модели, научились использовать Field для настройки валидации и обрабатывали вложенные модели. Также мы разобрались с обязательными и опциональными полями, кастомизировали ошибки и использовали Pydantic для валидации данных, как во входных запросах, так и в ответах API.

Сегодня мы свяжем всё это вместе и создадим рабочий пример полной валидации данных с использованием Pydantic. В рамках этого занятия мы возьмём модель с реальными сценариями и применим лучшие практики.

Постановка задачи: что мы реализуем

Представьте, что мы разрабатываем API для управления событиями (event management system). Это API позволяет нам:

• Создавать события (с информацией о названии, дате, месте и участниках).
• Получать информацию о событиях.
• Обновлять данные о событиях.
• Удалять события.

Мы будем учитывать:

• Валидацию входных данных для запросов.
• Структурирование данных с помощью вложенных моделей.
• Проверку ответов, чтобы убедиться, что наша система возвращает правильные данные.

Шаг 1: подготовка проекта

Если вы ещё не настроили проект FastAPI, выполните следующие команды для его инициализации. Убедитесь, что FastAPI и Uvicorn установлены:

pip install fastapi uvicorn

Создайте файл main.py, который будет нашим основным модулем.

Шаг 2: создание Pydantic-модели для события

Каждое событие, которое мы будем создавать, должно содержать следующую информацию:

• Название (title): обязательное текстовое поле.
• Дата (date): дата события — обязательно.
• Место проведения (location): строка с ограничением длины.
• Участники (participants): список участников, где каждый участник имеет имя и email.

Создадим Pydantic-модель для описания этого события. Вложим в неё модель участника.

from pydantic import BaseModel, Field, EmailStr
from datetime import date
from typing import List

# Модель для участника события
class Participant(BaseModel):
    name: str = Field(..., min_length=2, max_length=50, description="Имя участника")
    email: EmailStr = Field(..., description="Электронная почта участника")

# Модель для события
class Event(BaseModel):
    title: str = Field(..., min_length=1, max_length=100, description="Название события")
    date: date = Field(..., description="Дата события")
    location: str = Field(..., min_length=5, max_length=100, description="Место проведения")
    participants: List[Participant] = Field(..., description="Список участников события")

Что здесь происходит:

1. Мы используем Field, чтобы добавить ограничения и метаинформацию (например, минимум и максимум символов).
2. Поле email имеет тип EmailStr, чтобы автоматически проверять корректность email.
3. Поле participants ожидает список объектов типа Participant.

Шаг 3: настройка эндпоинтов для API

Теперь создадим эндпоинты FastAPI, которые позволят работать с событиями. Мы начнём с базового эндпоинта для создания нового события. Также добавим функцию для хранения данных.

Для простоты мы используем список в качестве временного хранилища. В реальных проектах для этого обычно используется база данных.

from fastapi import FastAPI, HTTPException

app = FastAPI()

# Временное "хранилище" для событий
event_storage = []

@app.post("/events/", response_model=Event, status_code=201)
async def create_event(event: Event):
    """
    Создание нового события.
    """
    # Проверяем, что событие с таким названием и датой не существует
    for existing_event in event_storage:
        if existing_event.title == event.title and existing_event.date == event.date:
            raise HTTPException(status_code=400, detail="Событие с таким названием и датой уже существует.")
    
    # Если всё ок, добавляем событие в "хранилище"
    event_storage.append(event)
    return event

Что здесь происходит:

1. Эндпоинт принимает объект типа Event (входные данные автоматически валидируются).
2. Мы проверяем, не существует ли события с таким же названием и датой.
3. Если проверка пройдена, мы добавляем событие в event_storage и возвращаем объект события в ответе.

Шаг 4: получение списка событий

Добавим эндпоинт для получения всех событий из нашего "хранилища".

@app.get("/events/", response_model=List[Event])
async def get_events():
    """
    Получение списка всех событий.
    """
    return event_storage

Этот эндпоинт возвращает список всех событий, хранящихся в event_storage. Благодаря параметру response_model, FastAPI автоматически проверяет, чтобы структура возвращаемых данных соответствовала модели List[Event].

Шаг 5: обновление данных о событии

Теперь добавим возможность обновления данных о событии. Для этого будем искать событие по его индексу.

from pydantic import BaseModel
from typing import Optional, List

# Модель для обновления события (все поля опциональны)
class EventUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    date: Optional[date]
    location: Optional[str] = Field(None, min_length=5, max_length=100)
    participants: Optional[List[Participant]]

@app.put("/events/{event_id}", response_model=Event)
async def update_event(event_id: int, event_update: EventUpdate):
    """
    Обновление данных о событии.
    """
    if event_id < 0 or event_id >= len(event_storage):
        raise HTTPException(status_code=404, detail="Событие не найдено.")
    
    # Получаем текущее событие
    current_event = event_storage[event_id]

    # Обновляем только переданные поля
    updated_event = current_event.copy(update=event_update.dict(exclude_unset=True))
    event_storage[event_id] = updated_event
    return updated_event

Что здесь происходит:

1. Мы используем модель EventUpdate, где все поля опциональны.
2. Параметр exclude_unset=True игнорирует отсутствующие поля в запросе, оставляя их без изменений.
3. Обновляем данные события только в случае успеха.

Шаг 6: удаление события

Добавим последний элемент CRUD — удаление события.

@app.delete("/events/{event_id}", status_code=204)
async def delete_event(event_id: int):
    """
    Удаление события по ID.
    """
    if event_id < 0 or event_id >= len(event_storage):
        raise HTTPException(status_code=404, detail="Событие не найдено.")
    
    # Удаляем событие
    event_storage.pop(event_id)

Шаг 7: тестирование приложения

Теперь мы можем протестировать наше приложение. Запустите сервер командой:

uvicorn main:app --reload

Примеры запросов:

1. Создать событие:

   
   POST /events/
   {
       "title": "Python Meetup",
       "date": "2023-12-01",
       "location": "Online",
       "participants": [
           {"name": "Алиса", "email": "alice@example.com"},
           {"name": "Боб", "email": "bob@example.com"}
       ]
   }
           

2. Получить все события:

   
   GET /events/
           

3. Обновить событие:

   
   PUT /events/0
   {
       "location": "New York",
       "participants": [{"name": "Чарли", "email": "charlie@example.com"}]
   }
           

4. Удалить событие:

   
   DELETE /events/0
           

Заключение

Мы создали полноценную систему валидации данных с помощью Pydantic и FastAPI. Этот пример показывает, как структурировать данные, обрабатывать вложенные модели и проверять корректность данных на всех этапах работы API. Используйте эти техники в своих проектах для повышения их надежности и удобства поддержки.

1

Задача

Модуль 4: FastAPI, 3 уровень, 9 лекция

Недоступна

Валидатор для проверки даты

Валидатор для проверки даты

Открыть

1

Задача

Модуль 4: FastAPI, 3 уровень, 9 лекция

Недоступна

Полноценный CRUD для событий

Полноценный CRUD для событий

Открыть

3

Опрос

Работа с обязательными и опциональными полями, 3 уровень, 9 лекция

Недоступен

Работа с обязательными и опциональными полями

Работа с обязательными и опциональными полями

Открыть