FastAPI — это современный, быстрый (высокопроизводительный) веб-фреймворк для создания API с Python 3.6+ на основе стандартных Python типов. Его основные преимущества — это простота, производительность и встроенная документация. Сегодня мы создадим простой, но функциональный Todolist с использованием FastAPI и базы данных SQLite. Это приложение поможет вам организовать ваши дела и станет отличным введением в мир веб-разработки с FastAPI.
Для начала, давайте подготовим нашу рабочую среду и структуру проекта.
Часть 1: Подготовка рабочей среды
Виртуальное окружение
Виртуальное окружение в Python — это изолированная среда, которая позволяет устанавливать пакеты и зависимости отдельно от глобальной системы. Это помогает избежать конфликтов между проектами и обеспечивает чистоту окружения. Вот как создать виртуальное окружение:
Создание виртуального окружения
Откройте терминал и выполните следующую команду:
python -m venv venv
Активация виртуального окружения
После создания, активируйте виртуальное окружение:
На Windows:
.\venv\Scripts\activate
На macOS/Linux:
source venv/bin/activate
Установка зависимостей
Убедитесь, что у вас установлен Python 3.6 или выше. Для начала установим необходимые зависимости:
pip install fastapi sqlalchemy pydantic uvicorn
- FastAPI — основной фреймворк для нашего API.
- SQLAlchemy — для работы с базой данных.
- Pydantic — для валидации данных.
- Uvicorn — ASGI сервер для запуска нашего приложения.
Структура проекта
Создайте папку для проекта, например, todolist, и внутри нее создайте следующие файлы:
todolist/
├── main.py
├── database.py
├── models.py
├── schemas.py
└── crud.py
Основные компоненты проекта
Теперь рассмотрим каждый файл подробно.
database.py
Этот файл будет содержать настройки для подключения к базе данных и создание сессий для работы с ней.
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Указываем URL базы данных
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# Создаем движок SQLAlchemy
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
# Создаем локальную сессию
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Базовый класс для моделей
Base = declarative_base()
models.py
Этот файл будет содержать определения моделей данных, которые будут отображаться в таблицы базы данных.
from sqlalchemy import Column, Integer, String, Boolean
from database import Base
# Определяем модель Task
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
completed = Column(Boolean, default=False)
- Task — модель задачи, которая будет храниться в таблице tasks.
- id — уникальный идентификатор задачи.
- title — заголовок задачи.
- description — описание задачи.
- completed — флаг, указывающий, завершена ли задача.
schemas.py
Этот файл будет содержать схемы (schemas) для валидации и сериализации данных.
from pydantic import BaseModel
# Базовая схема задачи
class TaskBase(BaseModel):
title: str
description: str | None = None
completed: bool | None = None
# Схема для создания задачи
class TaskCreate(TaskBase):
pass
# Схема для отображения задачи
class Task(TaskBase):
id: int
class Config:
orm_mode = True
- TaskBase — базовая схема для задачи.
- TaskCreate — схема для создания новой задачи.
- Task — схема для отображения задачи с дополнительным полем id.
Часть 2: Реализация CRUD операций
В первой части мы подготовили рабочую среду, определили базовые компоненты проекта, включая настройки базы данных, модели данных и схемы валидации. Во второй части мы реализуем основные операции CRUD (создание, чтение, обновление и удаление) с помощью FastAPI.
crud.py
В этом файле будут реализованы функции для выполнения CRUD операций.
from sqlalchemy.orm import Session
import models, schemas
# Функция для получения задачи по ID
def get_task(db: Session, task_id: int):
return db.query(models.Task).filter(models.Task.id == task_id).first()
# Функция для получения списка задач с возможностью пагинации
def get_tasks(db: Session, skip: int = 0, limit: int = 10):
return db.query(models.Task).offset(skip).limit(limit).all()
# Функция для создания новой задачи
def create_task(db: Session, task: schemas.TaskCreate):
db_task = models.Task(title=task.title, description=task.description, completed=task.completed)
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
# Функция для удаления задачи по ID
def delete_task(db: Session, task_id: int):
db_task = get_task(db, task_id)
if db_task:
db.delete(db_task)
db.commit()
return db_task
return None
- get_task — получает задачу по её ID.
- get_tasks — получает список задач с поддержкой пагинации (skip и limit).
- create_task — создает новую задачу и сохраняет её в базе данных.
- delete_task — удаляет задачу по её ID.
main.py
Теперь мы объединим все части вместе и создадим основные маршруты API в файле main.py.
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine
import models, schemas, crud
# Создание всех таблиц в базе данных
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Зависимость для получения сессии базы данных
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Маршрут для создания новой задачи
@app.post("/tasks/", response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
return crud.create_task(db=db, task=task)
# Маршрут для получения списка задач
@app.get("/tasks/", response_model=list[schemas.Task])
def read_tasks(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
tasks = crud.get_tasks(db, skip=skip, limit=limit)
return tasks
# Маршрут для получения задачи по ID
@app.get("/tasks/{task_id}", response_model=schemas.Task)
def read_task(task_id: int, db: Session = Depends(get_db)):
db_task = crud.get_task(db, task_id=task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
return db_task
# Маршрут для удаления задачи по ID
@app.delete("/tasks/{task_id}", response_model=schemas.Task)
def delete_task(task_id: int, db: Session = Depends(get_db)):
db_task = crud.delete_task(db=db, task_id=task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
return db_task
- create_task — маршрут для создания новой задачи.
- read_tasks — маршрут для получения списка задач с возможностью пагинации.
- read_task — маршрут для получения задачи по её ID.
- delete_task — маршрут для удаления задачи по её ID.
Часть 3: Запуск и тестирование приложения
В предыдущих частях мы настроили рабочую среду, определили модели данных, схемы и реализовали CRUD операции. В этой части мы запустим наше приложение, протестируем API и обсудим возможные улучшения.
Запуск приложения
Запуск сервера
Для запуска сервера используйте команду:
uvicorn main:app --reload
- main указывает на файл main.py и экземпляр FastAPI приложения app.
- —reload позволяет автоматически перезагружать сервер при изменениях в коде, что удобно для разработки.
Тестирование API
После запуска сервера API будет доступно по адресу http://127.0.0.1:8000.
FastAPI предоставляет встроенную документацию, которую можно открыть в браузере по адресу:
- Swagger UI: http://127.0.0.1:8000/docs
- ReDoc: http://127.0.0.1:8000/redoc
Создание новой задачи
Отправьте POST запрос на /tasks/ с телом запроса:
{
"title": "Прочитать книгу",
"description": "Закончить чтение книги по FastAPI",
"completed": false
}
Ответ будет содержать созданную задачу:
Получение списка задач
Отправьте GET запрос на /tasks/. Ответ будет содержать список задач:
[
{
"id": 1,
"title": "Прочитать книгу",
"description": "Закончить чтение книги по FastAPI",
"completed": false
}
]
Получение задачи по ID
Отправьте GET запрос на /tasks/1. Ответ будет содержать задачу с ID 1:
{
"id": 1,
"title": "Прочитать книгу",
"description": "Закончить чтение книги по FastAPI",
"completed": false
}
Удаление задачи
Отправьте DELETE запрос на /tasks/1. Ответ будет содержать удаленную задачу:
{
"id": 1,
"title": "Прочитать книгу",
"description": "Закончить чтение книги по FastAPI",
"completed": false
}
Возможные улучшения
Наше базовое приложение Todolist можно улучшить и расширить, добавив следующие возможности:
- Обновление задачи: Реализуйте маршрут для обновления существующей задачи.
- Авторизация и аутентификация: Добавьте аутентификацию пользователей, чтобы каждый пользователь мог управлять своими задачами.
- Более сложная логика задач: Добавьте функциональность для установки сроков выполнения задач, приоритетов и статусов.
- Тестирование: Напишите автоматические тесты для вашего API, используя фреймворки такие как pytest и httpx.
- Развертывание: Разверните ваше приложение на платформу, такую как Heroku, AWS, или DigitalOcean, чтобы сделать его доступным в интернете.
Комментарии
0