Приклад повної валідації даних за допомогою 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. Використовуйте ці техніки у своїх проєктах для підвищення їх надійності та зручності підтримки.