""" Task API endpoints. """ from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.crud import crud_task from app.db.session import get_db from app.models.task import TaskStatus from app.schemas.task import Task, TaskCreate, TaskUpdate router = APIRouter() @router.get("", response_model=List[Task]) def read_tasks( db: Session = Depends(get_db), status: Optional[TaskStatus] = None, title: Optional[str] = None, skip: int = 0, limit: int = 100, ) -> Any: """ Retrieve tasks. Args: db: Database session. status: Filter by task status. title: Filter by task title (partial match). skip: Number of records to skip. limit: Maximum number of records to return. Returns: List[Task]: List of tasks. """ if status: return crud_task.task.get_by_status(db, status=status, skip=skip, limit=limit) elif title: return crud_task.task.get_by_title(db, title=title, skip=skip, limit=limit) return crud_task.task.get_multi(db, skip=skip, limit=limit) @router.post("", response_model=Task, status_code=status.HTTP_201_CREATED) def create_task( *, db: Session = Depends(get_db), task_in: TaskCreate, ) -> Any: """ Create new task. Args: db: Database session. task_in: Task data. Returns: Task: Created task. """ task = crud_task.task.create(db=db, obj_in=task_in) return task @router.get("/{id}", response_model=Task) def read_task( *, db: Session = Depends(get_db), id: int, ) -> Any: """ Get task by ID. Args: db: Database session. id: Task ID. Returns: Task: Task with specified ID. Raises: HTTPException: If task not found. """ task = crud_task.task.get(db=db, id=id) if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found", ) return task @router.put("/{id}", response_model=Task) def update_task( *, db: Session = Depends(get_db), id: int, task_in: TaskUpdate, ) -> Any: """ Update a task. Args: db: Database session. id: Task ID. task_in: Task update data. Returns: Task: Updated task. Raises: HTTPException: If task not found. """ task = crud_task.task.get(db=db, id=id) if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found", ) task = crud_task.task.update(db=db, db_obj=task, obj_in=task_in) return task @router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def delete_task( *, db: Session = Depends(get_db), id: int, ) -> None: """ Delete a task. Args: db: Database session. id: Task ID. Raises: HTTPException: If task not found. """ task = crud_task.task.get(db=db, id=id) if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Task not found", ) crud_task.task.remove(db=db, id=id) return None