crud fastapicrud fastapi

FastAPI — это современный и быстрый фреймворк для создания веб-приложений и API на Python. Он идеально подходит для создания приложений с CRUD-функционалом (создание, чтение, обновление, удаление данных). В этом руководстве мы шаг за шагом разберем, как создать простое CRUD-приложение на FastAPI.

Установка и настройка проекта

Для начала создадим новый проект и установим необходимые зависимости.

  1. Создайте виртуальное окружение: python3 -m venv venv source venv/bin/activate # активация на MacOS/Linux venv\Scripts\activate # активация на Windows
  2. Установите FastAPI и Uvicorn — сервер для запуска нашего приложения. pip install fastapi uvicorn

Теперь у нас готово окружение, и мы можем начать писать код.

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

Чтобы проект был удобным и структурированным, создадим папку app, в которой будут храниться файлы приложения, а также файл main.py для запуска.

my_fastapi_crud/

├── app/
│ ├── models.py
│ ├── schemas.py
│ ├── database.py
│ └── crud.py

└── main.py

Настройка базы данных

Для простоты мы будем использовать SQLite, но структура останется аналогичной и для других баз данных, таких как PostgreSQL или MySQL. Нам потребуется SQLAlchemy для работы с базой данных, поэтому установим его:

pip install sqlalchemy

Создание модели базы данных

Начнем с создания модели данных. В FastAPI модели хранятся в отдельном файле, обычно это models.py. В этом файле мы создадим модель для базы данных с использованием SQLAlchemy.

Создадим файл app/models.py:

# app/models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String, index=True)
price = Column(Integer)
available = Column(String, default="Yes")

Модель Item описывает таблицу items с полями id, name, description, price, и available.

Настройка подключения к базе данных

Теперь создадим файл database.py, в котором будет логика подключения к базе данных и создание сессий для выполнения операций.

Создадим app/database.py:

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" # путь к нашей базе данных SQLite

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Создание таблиц в базе данных
Base.metadata.create_all(bind=engine)

Здесь мы настраиваем подключение к базе данных SQLite. SessionLocal создаёт сессии для выполнения операций с базой данных, а Base.metadata.create_all инициализирует таблицу items в базе данных.

Создание схем данных (Schemas)

Схемы описывают структуру данных, которую принимает и возвращает наш API. Они помогают задать формат входных и выходных данных и делают приложение более безопасным.

Создадим app/schemas.py:

# app/schemas.py
from pydantic import BaseModel

class ItemBase(BaseModel):
name: str
description: str
price: int
available: str

class ItemCreate(ItemBase):
pass

class Item(ItemBase):
id: int

class Config:
orm_mode = True
  • ItemBase — базовая схема с основными полями.
  • ItemCreate — схема для создания новых элементов, наследует ItemBase.
  • Item — схема, представляющая элемент с полем id, используется для отображения элементов из базы данных.

Логика CRUD

Теперь создадим функции для реализации операций CRUD. Эти функции будут отвечать за добавление, получение, обновление и удаление элементов из базы данных.

Создадим app/crud.py:

# app/crud.py
from sqlalchemy.orm import Session
from app import models, schemas

def get_item(db: Session, item_id: int):
return db.query(models.Item).filter(models.Item.id == item_id).first()

def get_items(db: Session, skip: int = 0, limit: int = 10):
return db.query(models.Item).offset(skip).limit(limit).all()

def create_item(db: Session, item: schemas.ItemCreate):
db_item = models.Item(name=item.name, description=item.description, price=item.price, available=item.available)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item

def update_item(db: Session, item_id: int, item: schemas.ItemCreate):
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
db_item.name = item.name
db_item.description = item.description
db_item.price = item.price
db_item.available = item.available
db.commit()
db.refresh(db_item)
return db_item

def delete_item(db: Session, item_id: int):
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
db.delete(db_item)
db.commit()
return db_item

Функции выполняют запросы к базе данных через SQLAlchemy для получения, добавления, обновления и удаления записей.

Основные маршруты приложения

Теперь свяжем нашу логику CRUD с маршрутами API, чтобы пользователи могли отправлять запросы и взаимодействовать с приложением. Создадим основной файл приложения main.py:

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from app import models, schemas, crud
from app.database import SessionLocal, engine

# Создаем экземпляр FastAPI
app = FastAPI()

# Зависимость для получения сессии базы данных
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

# Создание нового элемента
@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
return crud.create_item(db=db, item=item)

# Получение всех элементов
@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
return crud.get_items(db=db, skip=skip, limit=limit)

# Получение элемента по ID
@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
db_item = crud.get_item(db=db, item_id=item_id)
if db_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return db_item

# Обновление элемента по ID
@app.put("/items/{item_id}", response_model=schemas.Item)
def update_item(item_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
db_item = crud.update_item(db=db, item_id=item_id, item=item)
if db_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return db_item

# Удаление элемента по ID
@app.delete("/items/{item_id}", response_model=schemas.Item)
def delete_item(item_id: int, db: Session = Depends(get_db)):
db_item = crud.delete_item(db=db, item_id=item_id)
if db_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return db_item

Здесь мы создаем основные маршруты для выполнения CRUD-операций:

  • POST /items/ — добавляет новый элемент.
  • GET /items/ — возвращает список всех элементов.
  • GET /items/{item_id} — возвращает элемент по item_id.
  • PUT /items/{item_id} — обновляет элемент по item_id.
  • DELETE /items/{item_id} — удаляет элемент по item_id.

Запуск приложения

Теперь, когда код готов, запустим приложение с помощью Uvicorn:

uvicorn main:app --reload

Приложение будет доступно по адресу http://127.0.0.1:8000. Чтобы протестировать API, откройте браузер и перейдите на страницу документации Swagger: http://127.0.0.1:8000/docs.

Заключение

В этом руководстве мы создали простое CRUD-приложение на FastAPI, включающее подключение к базе данных, модели, схемы и маршруты для создания, чтения, обновления и удаления записей.

Поделится