from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.orm import Session from datetime import datetime, date from app import schemas from app.models.todo import Todo from app.db.session import get_db router = APIRouter() @router.get("/", response_model=List[schemas.Todo]) def read_todos( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, ) -> Any: """ Retrieve todos. """ todos = db.query(Todo).order_by(Todo.created_at.desc()).offset(skip).limit(limit).all() return todos @router.post("/", response_model=schemas.Todo, status_code=status.HTTP_201_CREATED) def create_todo( *, db: Session = Depends(get_db), todo_in: schemas.TodoCreate, ) -> Any: """ Create new todo. """ todo = Todo( title=todo_in.title, description=todo_in.description, completed=todo_in.completed, due_date=todo_in.due_date, ) db.add(todo) db.commit() db.refresh(todo) return todo @router.get("/{id}", response_model=schemas.Todo) def read_todo( *, db: Session = Depends(get_db), id: int, ) -> Any: """ Get todo by ID. """ todo = db.query(Todo).filter(Todo.id == id).first() if not todo: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Todo with ID {id} not found" ) return todo @router.put("/{id}", response_model=schemas.Todo) def update_todo( *, db: Session = Depends(get_db), id: int, todo_in: schemas.TodoUpdate, ) -> Any: """ Update a todo. """ todo = db.query(Todo).filter(Todo.id == id).first() if not todo: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Todo with ID {id} not found" ) update_data = todo_in.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(todo, field, value) db.add(todo) db.commit() db.refresh(todo) return todo @router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_todo( *, db: Session = Depends(get_db), id: int, ) -> Any: """ Delete a todo. """ todo = db.query(Todo).filter(Todo.id == id).first() if not todo: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Todo with ID {id} not found" ) db.delete(todo) db.commit() return None @router.get("/due-date/", response_model=List[schemas.Todo]) def get_todos_by_due_date( *, db: Session = Depends(get_db), start_date: Optional[date] = Query(None, description="Filter todos due on or after this date"), end_date: Optional[date] = Query(None, description="Filter todos due on or before this date"), skip: int = 0, limit: int = 100, ) -> Any: """ Filter todos by due date range. """ query = db.query(Todo) if start_date: # Convert date to datetime with time at beginning of day start_datetime = datetime.combine(start_date, datetime.min.time()) query = query.filter(Todo.due_date >= start_datetime) if end_date: # Convert date to datetime with time at end of day end_datetime = datetime.combine(end_date, datetime.max.time()) query = query.filter(Todo.due_date <= end_datetime) # Only return todos with due dates if filtering by date if start_date or end_date: query = query.filter(Todo.due_date.isnot(None)) todos = query.order_by(Todo.due_date.asc()).offset(skip).limit(limit).all() return todos