from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app import crud from app.db.session import get_db from app.schemas.task import Task, TaskCreate, TaskUpdate router = APIRouter() @router.get("/", response_model=List[Task]) def read_tasks( is_completed: Optional[bool] = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db), ): """ Retrieve tasks with optional filtering. """ tasks = crud.task.get_tasks(db, skip=skip, limit=limit, is_completed=is_completed) return tasks @router.post("/", response_model=Task) def create_task( task_in: TaskCreate, db: Session = Depends(get_db), ): """ Create new task. """ task = crud.task.create_task(db=db, task_in=task_in) return task @router.get("/{task_id}", response_model=Task) def read_task( task_id: int, db: Session = Depends(get_db), ): """ Get task by ID. """ task = crud.task.get_task(db=db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task @router.put("/{task_id}", response_model=Task) def update_task( task_id: int, task_in: TaskUpdate, db: Session = Depends(get_db), ): """ Update a task. """ task = crud.task.get_task(db=db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") task = crud.task.update_task(db=db, task=task, task_in=task_in) return task @router.delete("/{task_id}", response_model=Task) def delete_task( task_id: int, db: Session = Depends(get_db), ): """ Delete a task. """ task = crud.task.get_task(db=db, task_id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") task_data = Task.from_orm(task) crud.task.delete_task(db=db, task=task) return task_data