fastapi-todolist/main.py

from typing import Self

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, field_validator, model_validator
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError

from dependencies import SessionDep
from models import Todo

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class todocreate(BaseModel):
    task: str = Field(min_length=1, max_length=200)

    @field_validator("task")
    @classmethod
    def task_must_not_be_blank(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("task must not be blank")

        return value


class todoupdate(BaseModel):
    completion: bool | None = None
    task: str | None = Field(
        default=None,
        min_length=1,
        max_length=200,
    )

    @field_validator("task")
    @classmethod
    def task_must_not_be_blank(
        cls,
        value: str | None,
    ) -> str | None:
        if value is not None and not value.strip():
            raise ValueError("task must not be blank")

        return value

    @model_validator(mode="after")
    def at_least_one_field_must_be_provided(self) -> Self:
        if self.task is None and self.completion is None:
            raise ValueError("task or completion must be provided")

        return self


class todoresponse(BaseModel):
    id: int
    task: str
    completion: bool


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/todos", response_model=list[todoresponse])
def todos_get(session: SessionDep):
    statement = select(Todo)
    todos = session.scalars(statement).all()
    return todos


@app.post("/todos", response_model=todoresponse)
def todos_create(
    todo_data: todocreate,
    session: SessionDep,
):
    try:
        new_todo = Todo(task=todo_data.task)
        session.add(new_todo)
        session.commit()
        return new_todo
    except SQLAlchemyError:
        session.rollback()
        raise


@app.patch("/todos/{todo_id}", response_model=todoresponse)
def todos_patch(
    todo_id: int,
    update_data: todoupdate,
    session: SessionDep,
):
    try:
        todo = session.get(Todo, todo_id)
        if todo is None:
            raise HTTPException(status_code=404, detail="Todo not found")
        if update_data.task is not None:
            todo.task = update_data.task
        if update_data.completion is not None:
            todo.completion = update_data.completion
        session.commit()
        return todo
    except SQLAlchemyError:
        session.rollback()
        raise


@app.delete("/todos/{todo_id}", response_model=todoresponse)
def todos_delete(
    todo_id: int,
    session: SessionDep,
):
    try:
        todo = session.get(Todo, todo_id)
        if todo is None:
            raise HTTPException(status_code=404, detail="Todo not found")
        session.delete(todo)
        session.commit()
        return todo
    except SQLAlchemyError:
        session.rollback()
        raise