Создание REST API с помощью FastAPI и SQLite

Для разработки современного REST API на Python часто выбирают фреймворк FastAPI. Он поддерживает асинхронность, автоматическую документацию и валидацию данных. Рассмотрим создание простого API для управления задачами (TODO).

Шаг 1. Установка зависимостей

Необходимо установить FastAPI, сервер Uvicorn и SQLAlchemy для работы с базой данных:

pip install fastapi uvicorn sqlalchemy

Python пример разработки (пример разработки на python)

Шаг 2. Создание структуры проекта

Создадим файл main.py со следующим содержимым:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Настройка базы данных SQLite
SQLALCHEMY_DATABASE_URL = 'sqlite:///./todos.db'
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

# Определение модели Task
class Task(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    completed = Column(Boolean, default=False)

Base.metadata.create_all(bind=engine)

# Pydantic модель для запросов и ответов
class TaskCreate(BaseModel):
    title: str
    completed: bool = False

class TaskResponse(BaseModel):
    id: int
    title: str
    completed: bool

app = FastAPI()

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

@app.post('/tasks/', response_model=TaskResponse)
def create_task(task: TaskCreate):
    db = next(get_db())
    db_task = Task(title=task.title, completed=task.completed)
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

@app.get('/tasks/', response_model=list[TaskResponse])
def read_tasks():
    db = next(get_db())
    tasks = db.query(Task).all()
    return tasks

@app.get('/tasks/{task_id}', response_model=TaskResponse)
def read_task(task_id: int):
    db = next(get_db())
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        raise HTTPException(status_code=404, detail='Task not found')
    return task

@app.put('/tasks/{task_id}', response_model=TaskResponse)
def update_task(task_id: int, task: TaskCreate):
    db = next(get_db())
    db_task = db.query(Task).filter(Task.id == task_id).first()
    if not db_task:
        raise HTTPException(status_code=404, detail='Task not found')
    db_task.title = task.title
    db_task.completed = task.completed
    db.commit()
    db.refresh(db_task)
    return db_task

@app.delete('/tasks/{task_id}')
def delete_task(task_id: int):
    db = next(get_db())
    db_task = db.query(Task).filter(Task.id == task_id).first()
    if not db_task:
        raise HTTPException(status_code=404, detail='Task not found')
    db.delete(db_task)
    db.commit()
    return {'ok': True}
  

Шаг 3. Запуск сервера

Выполните команду в терминале:

uvicorn main:app --reload

Сервер запустится на http://127.0.0.1:8000. Документация доступна по адресу /docs.