PR4 - master - ViliaIbatullina/HTML

PR4
122 строки4 KB
ViliaIbatullina
practice_4
19 апр 2025, 07:02
19 апр 2025, 07:02d895e8e
├── database.py         # Настройка базы данных и миграции  
├── main.py             # Основной файл FastAPI-приложения  
├── model/              # Модели данных  
│   ├── __init__.py     # Инициализация пакета model
│   └── todo.py         # Описание модели Todo  
├── repository/         # Логика работы с базой данных  
│   ├── __init__.py     # Инициализация пакета repository
│   └── todo.py         # CRUD-операции для Todo  
└── todos.db            # Файл базы данных SQLite  

from sqlalchemy import Column, Integer, String, Boolean
from database import Base
from pydantic import BaseModel

class TodoModel(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    completed = Column(Boolean, default=False)

class TodoInput(BaseModel):
    title: str
    completed: bool = False

class Todo(TodoInput):
    id: int

    from sqlalchemy.orm import Session
from model.todo import TodoModel, TodoInput, Todo

class TodoRepository:
    @staticmethod
    def get_all(db: Session):
        return db.query(TodoModel).all()

    @staticmethod
    def get_by_id(db: Session, todo_id: int):
        return db.query(TodoModel).filter(TodoModel.id == todo_id).first()

    @staticmethod
    def create(db: Session, todo_data: TodoInput):
        new_todo = TodoModel(title=todo_data.title, completed=todo_data.completed)
        db.add(new_todo)
        db.commit()
        db.refresh(new_todo)
        return Todo(id=new_todo.id, title=new_todo.title, completed=new_todo.completed)

    @staticmethod
    def update(db: Session, todo_id: int, todo_data: TodoInput):
        todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
        if todo is None:
            return None
        todo.title = todo_data.title
        todo.completed = todo_data.completed
        db.commit()
        db.refresh(todo)
        return Todo(id=todo.id, title=todo.title, completed=todo.completed)

    @staticmethod
    def delete(db: Session, todo_id: int):
        todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
        if todo is None:
            return False
        db.delete(todo)
        db.commit()
        return True

        from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./todos.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def migrate():
    Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from repository.todo import TodoRepository
from model.todo import TodoInput, Todo
from database import get_db, migrate

app = FastAPI()

# Выполняем миграцию при запуске приложения
migrate()

@app.get("/todos", response_model=list[Todo])
def get_todos(db: Session = Depends(get_db)):
    return TodoRepository.get_all(db)

@app.post("/todos", response_model=Todo, status_code=201)
def create_todo(todo: TodoInput, db: Session = Depends(get_db)):
    return TodoRepository.create(db, todo)

@app.get("/todos/{id}", response_model=Todo)
def get_todo(id: int, db: Session = Depends(get_db)):
    todo = TodoRepository.get_by_id(db, id)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.put("/todos/{id}", response_model=Todo)
def update_todo(id: int, todo_data: TodoInput, db: Session = Depends(get_db)):
    todo = TodoRepository.update(db, id, todo_data)
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.delete("/todos/{id}", status_code=204)
def delete_todo(id: int, db: Session = Depends(get_db)):
    if not TodoRepository.delete(db, id):
        raise HTTPException(status_code=404, detail="Todo not found")