CRUD — это аббревиатура, которая расшифровывается как Create, Read, Update и Delete, то есть "Создать, Прочитать, Обновить, Удалить". Все эти операции лежат в основе каждого приложения, работающего с данными. Хотите записать пользователя в базу? Это Create. Нужно показать список товаров? Read. А вдруг пользователь решил обновить свой номер телефона? Update. Ну, а удаление данных (иногда с грустью, как прощание) — это Delete.
На собеседованиях вакансий на позицию Python-разработчика вас обязательно спросят про реализацию CRUD API. В реальной жизни такие API используются для управления объектами, взаимодействия с базой данных и построения надежных систем. На практике API, которые мы создаем, почти всегда состоят из подобных операций.
Сегодня мы создадим мини-приложение на FastAPI, которое будет управлять списком книг. Каждая книга будет характеризоваться следующими атрибутами:
id— уникальный идентификатор книги (целочисленный).title— название книги (строка).author— автор (строка).year— год публикации (целое число).
Настройка проекта
Для начала убедитесь, что у вас установлен FastAPI, Uvicorn и Pydantic. Если нет, то исправим это командой:
pip install fastapi uvicorn pydantic
Создайте структуру проекта:
crud_app/
├── main.py
└── models.py
В файле main.py будет описана логика нашего приложения, а models.py — место, где мы определим наши Pydantic-модели.
Моделирование данных
Начнем с определения книги как объекта. Откройте models.py и создайте следующее:
from pydantic import BaseModel
from typing import Optional
# Модель данных для книги
class Book(BaseModel):
id: int
title: str
author: str
year: int
# Модель для обновления книги (частичное обновление)
class BookUpdate(BaseModel):
title: Optional[str] = None
author: Optional[str] = None
year: Optional[int] = None
Мы определили две модели:
Book— полная модель книги.BookUpdate— позволяет обновлять только часть полей объекта.
Реализация CRUD-операций
Теперь переходим к главному файлу main.py. Начнем с подключения базовых библиотек и инициализации FastAPI:
from fastapi import FastAPI, HTTPException
from models import Book, BookUpdate
app = FastAPI()
# Хранилище данных
books = [] # В данной реализации используем список для хранения книг.
Создание книги (Create)
Мы создадим эндпоинт, который принимает данные книги в формате JSON, валидирует их и добавляет книгу в список.
@app.post("/books/", response_model=Book)
async def create_book(book: Book):
# Проверяем, есть ли уже книга с таким id
for existing_book in books:
if existing_book.id == book.id:
raise HTTPException(status_code=400, detail="Книга с таким ID уже существует")
books.append(book)
return book
Получение списка книг (Read)
Теперь сделаем эндпоинт для чтения списка всех книг.
@app.get("/books/", response_model=list[Book])
async def get_books():
return books
Или отдельной книги по id:
@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: int):
for book in books:
if book.id == book_id:
return book
raise HTTPException(status_code=404, detail="Книга не найдена")
Обновление книги (Update)
Обновление книги будет обрабатываться с помощью PUT-запроса, и частичное обновление будет реализовано с использованием модели BookUpdate.
@app.put("/books/{book_id}", response_model=Book)
async def update_book(book_id: int, book_update: BookUpdate):
for book in books:
if book.id == book_id:
# Обновляем только те поля, которые были переданы
if book_update.title is not None:
book.title = book_update.title
if book_update.author is not None:
book.author = book_update.author
if book_update.year is not None:
book.year = book_update.year
return book
raise HTTPException(status_code=404, detail="Книга не найдена")
Удаление книги (Delete)
Для удаления книги также добавим эндпоинт:
@app.delete("/books/{book_id}", response_model=dict)
async def delete_book(book_id: int):
for book in books:
if book.id == book_id:
books.remove(book)
return {"message": "Книга успешно удалена"}
raise HTTPException(status_code=404, detail="Книга не найдена")
Валидация данных и обработка ошибок
FastAPI уже автоматически делает много работы за нас благодаря Pydantic. Если данные не будут соответствовать модели, API вернет понятное сообщение об ошибке. Пример:
{
"detail": [
{
"loc": ["body", "title"],
"msg": "field required",
"type": "value_error.missing"
}
]
}
В случае пользовательских ошибок, таких как попытка создать книгу с уже существующим ID, используем HTTPException.
Быстрая интеграция с базой данных (опционально)
На данный момент наши данные хранятся в памяти (список books). Это удобно для быстрых экспериментов, но недопустимо для реальных проектов. Подключим базу данных SQLite.
Установите SQLAlchemy и aiosqlite:
pip install sqlalchemy aiosqlite
Вынесем модели в models.py и создадим SQLAlchemy ORM:
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
# Модель книги для работы с базой данных
class BookDB(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
author = Column(String)
year = Column(Integer)
# Создаем SQLite базу данных
DATABASE_URL = "sqlite:///./books.db"
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Теперь в CRUD-эндпоинтах вы будете использовать SessionLocal для взаимодействия с базой. Например, вместо добавления в список books используйте базовый SQL-запрос:
db = SessionLocal()
db.add(new_book)
db.commit()
Этот шаг выводит вашу реализацию на новый уровень, позволяя работать с большими объемами данных.
Вот и всё! Мы создали полноценный CRUD API, который можно дальше улучшать и адаптировать под нужды вашего проекта. Это мощный старт для построения ваших приложений!
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ