CRUD API на FastAPI, Pydantic v2 і JWT

Вступ

FastAPI давно став одним із найзручніших фреймворків для створення швидких API на Python. Він поєднує високу продуктивність, автоматичну генерацію документації та простий синтаксис. Якщо додати до нього Pydantic v2 для валідації даних і JWT для автентифікації, можна отримати сучасний каркас для будь-якого вебсервісу.

У цьому матеріалі створимо простий, але повноцінний CRUD API для сутності Task. Ми реалізуємо створення, читання, оновлення та видалення записів, а також захистимо частину ендпоїнтів токеном JWT.

Що будемо будувати

Наш приклад буде умовним сервісом для керування задачами. Користувач зможе:

• зареєструватися та отримати JWT токен;
• увійти в систему;
• створювати задачі;
• отримувати список задач;
• переглядати одну задачу за ID;
• оновлювати задачу;
• видаляти задачу.

Для простоти ми використаємо in-memory сховище, тобто дані зберігатимуться в пам’яті програми. Це хороший варіант для навчального прикладу. Для реального проєкту краще підключити PostgreSQL або іншу базу даних.

Структура проєкту

Зручніше одразу розділити код на кілька файлів. Приклад структури:

• main.py — точка входу;
• schemas.py — Pydantic-моделі;
• auth.py — логіка JWT;
• storage.py — тимчасове сховище задач;
• routes.py — маршрути API.

Такий поділ допомагає не змішувати логіку автентифікації, валідації та обробки запитів в одному файлі.

Pydantic v2: схеми даних

У Pydantic v2 зручно описувати моделі для запиту та відповіді. Окремі схеми дозволяють контролювати, які поля приймає API і що саме повертає клієнту.

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=6)

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

class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    description: Optional[str] = None

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=1, max_length=100)
    description: Optional[str] = None
    is_done: Optional[bool] = None

class TaskRead(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    is_done: bool

У Pydantic v2 варто звертати увагу на чіткість типів і обмеження через Field. Це зменшує кількість помилок на рівні бізнес-логіки, бо некоректні дані не потрапляють далі в застосунок.

JWT-автентифікація

JWT зручний для API, де не потрібно зберігати серверну сесію. Після логіну користувач отримує токен, який передає в заголовку Authorization: Bearer .... Сервер перевіряє підпис токена та вирішує, чи можна відкрити доступ до захищеного маршруту.

Приклад логіки токенів

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET_KEY = "super-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

У реальному проєкті секретний ключ слід зберігати в змінних середовища, а не в коді. Це базова практика безпеки для будь-якого Python API.

Пам’ятати користувачів і задачі

Для навчального прикладу використаємо прості словники. Один зберігатиме користувачів, інший — задачі.

users_db = {}
tasks_db = {}
next_task_id = 1

У production-версії цю частину замінюють ORM, наприклад SQLAlchemy, та реальну базу даних. Але сам принцип CRUD залишиться таким самим.

Маршрути FastAPI

Тепер зберемо основну логіку API. Спочатку створимо ендпоїнти для реєстрації та логіну, потім — захищені маршрути для задач.

from fastapi import FastAPI, HTTPException, Depends, Header
from schemas import UserCreate, UserLogin, TokenResponse, TaskCreate, TaskUpdate, TaskRead
from auth import create_access_token, verify_token

app = FastAPI(title="CRUD API with JWT")

@app.post("/register", response_model=TokenResponse)
def register(user: UserCreate):
    if user.email in users_db:
        raise HTTPException(status_code=400, detail="Користувач уже існує")
    users_db[user.email] = {"password": user.password}
    token = create_access_token({"sub": user.email})
    return {"access_token": token, "token_type": "bearer"}

@app.post("/login", response_model=TokenResponse)
def login(data: UserLogin):
    user = users_db.get(data.email)
    if not user or user["password"] != data.password:
        raise HTTPException(status_code=401, detail="Невірні облікові дані")
    token = create_access_token({"sub": data.email})
    return {"access_token": token, "token_type": "bearer"}

Для захисту ендпоїнтів додамо допоміжну функцію, яка читає токен із заголовка. Якщо токен невалідний, API поверне помилку доступу.

def get_current_user(authorization: str = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Відсутній токен")
    token = authorization.split(" ", 1)[1]
    payload = verify_token(token)
    if not payload:
        raise HTTPException(status_code=401, detail="Недійсний токен")
    return payload["sub"]

CRUD для задач

Після цього реалізуємо повний набір операцій.

@app.post("/tasks", response_model=TaskRead)
def create_task(task: TaskCreate, current_user: str = Depends(get_current_user)):
    global next_task_id
    new_task = {
        "id": next_task_id,
        "title": task.title,
        "description": task.description,
        "is_done": False,
        "owner": current_user,
    }
    tasks_db[next_task_id] = new_task
    next_task_id += 1
    return new_task

@app.get("/tasks", response_model=list[TaskRead])
def list_tasks(current_user: str = Depends(get_current_user)):
    return [task for task in tasks_db.values() if task["owner"] == current_user]

@app.get("/tasks/{task_id}", response_model=TaskRead)
def get_task(task_id: int, current_user: str = Depends(get_current_user)):
    task = tasks_db.get(task_id)
    if not task or task["owner"] != current_user:
        raise HTTPException(status_code=404, detail="Задачу не знайдено")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskRead)
def update_task(task_id: int, data: TaskUpdate, current_user: str = Depends(get_current_user)):
    task = tasks_db.get(task_id)
    if not task or task["owner"] != current_user:
        raise HTTPException(status_code=404, detail="Задачу не знайдено")
    if data.title is not None:
        task["title"] = data.title
    if data.description is not None:
        task["description"] = data.description
    if data.is_done is not None:
        task["is_done"] = data.is_done
    return task

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, current_user: str = Depends(get_current_user)):
    task = tasks_db.get(task_id)
    if not task or task["owner"] != current_user:
        raise HTTPException(status_code=404, detail="Задачу не знайдено")
    del tasks_db[task_id]
    return {"detail": "Задачу видалено"}

Чому це працює добре

FastAPI спрощує роботу з залежностями, тому перевірка JWT інтегрується природно через Depends. Pydantic v2 забезпечує сувору валідацію вхідних даних і зрозумілі схеми відповіді. У результаті код виходить коротким, але структурованим.

• автентифікація відокремлена від CRUD-логіки;
• моделі чітко описують контракт API;
• маршрути легко тестувати й розширювати;
• архітектуру можна безболісно перенести на базу даних.

Що покращити далі

Після цього навчального прикладу наступним кроком зазвичай стає підключення PostgreSQL, хешування паролів, винесення конфігурації в .env та додавання тестів. Також корисно реалізувати ролі користувачів, pagination для списків і централізовану обробку помилок.

Якщо вам потрібно швидко стартувати з сучасним Python API, FastAPI + Pydantic v2 + JWT — дуже вдале поєднання. Воно дає чистий код, передбачувану структуру та хороший фундамент для подальшого розвитку проєкту.

Comments

comments