from typing import List, Optional from datetime import datetime, timedelta from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from sqlalchemy import and_, not_ from app.db.session import get_db from app.models.todo import Todo from app.schemas.todo import TodoCreate, TodoUpdate, Todo as TodoSchema, PriorityEnum as SchemaPriorityEnum router = APIRouter( prefix="/todos", tags=["todos"], responses={404: {"description": "Not found"}}, ) @router.post("/", response_model=TodoSchema, status_code=status.HTTP_201_CREATED) def create_todo(todo: TodoCreate, db: Session = Depends(get_db)): db_todo = Todo(**todo.dict()) db.add(db_todo) db.commit() db.refresh(db_todo) return db_todo @router.get("/", response_model=List[TodoSchema]) def read_todos( skip: int = 0, limit: int = 100, completed: Optional[bool] = None, priority: Optional[SchemaPriorityEnum] = None, due_date_before: Optional[datetime] = None, due_date_after: Optional[datetime] = None, due_soon: Optional[bool] = None, overdue: Optional[bool] = None, db: Session = Depends(get_db) ): query = db.query(Todo) filters = [] # Apply completed filter if completed is not None: filters.append(Todo.completed == completed) # Apply priority filter if priority is not None: filters.append(Todo.priority == priority.value) # Apply due date filters if due_date_before is not None: filters.append(Todo.due_date <= due_date_before) if due_date_after is not None: filters.append(Todo.due_date >= due_date_after) # Due soon filter (next 3 days) if due_soon is True: now = datetime.now() three_days_later = now + timedelta(days=3) filters.append(and_( Todo.due_date >= now, Todo.due_date <= three_days_later, not_(Todo.completed) )) # Overdue filter if overdue is True: now = datetime.now() filters.append(and_( Todo.due_date < now, not_(Todo.completed) )) # Apply all filters if filters: query = query.filter(and_(*filters)) # Order by priority (high to low) and due date (closest first) query = query.order_by( # Custom ordering for priority # This works because of our string values 'high', 'medium', 'low' Todo.priority.desc(), # Due date, null values last Todo.due_date.asc() ) todos = query.offset(skip).limit(limit).all() return todos @router.get("/{todo_id}", response_model=TodoSchema) def read_todo(todo_id: int, db: Session = Depends(get_db)): db_todo = db.query(Todo).filter(Todo.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") return db_todo @router.put("/{todo_id}", response_model=TodoSchema) def update_todo(todo_id: int, todo: TodoUpdate, db: Session = Depends(get_db)): db_todo = db.query(Todo).filter(Todo.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") update_data = todo.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_todo, key, value) db.commit() db.refresh(db_todo) return db_todo @router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_todo(todo_id: int, db: Session = Depends(get_db)): db_todo = db.query(Todo).filter(Todo.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") db.delete(db_todo) db.commit() return None