62cd6ed746
- Add complete CRUD operations for task management - Implement task filtering by status, priority, and search - Add task statistics endpoint for summary data - Configure SQLite database with Alembic migrations - Set up FastAPI with CORS support and API documentation - Include health check endpoint and base URL information - Add comprehensive README with API usage examples - Structure project with proper separation of concerns
73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
from app.crud import task as crud_task
|
|
from app.schemas.task import TaskCreate, TaskUpdate, TaskResponse
|
|
from app.models.task import TaskStatus, TaskPriority
|
|
from app.db.session import get_db
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[TaskResponse])
|
|
def read_tasks(
|
|
skip: int = Query(0, ge=0, description="Number of tasks to skip"),
|
|
limit: int = Query(100, ge=1, le=1000, description="Maximum number of tasks to return"),
|
|
status: Optional[TaskStatus] = Query(None, description="Filter by task status"),
|
|
priority: Optional[TaskPriority] = Query(None, description="Filter by task priority"),
|
|
search: Optional[str] = Query(None, description="Search in title and description"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
tasks = crud_task.get_tasks(
|
|
db=db,
|
|
skip=skip,
|
|
limit=limit,
|
|
status=status,
|
|
priority=priority,
|
|
search=search
|
|
)
|
|
return tasks
|
|
|
|
|
|
@router.post("/", response_model=TaskResponse)
|
|
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
|
|
return crud_task.create_task(db=db, task=task)
|
|
|
|
|
|
@router.get("/{task_id}", response_model=TaskResponse)
|
|
def read_task(task_id: int, db: Session = Depends(get_db)):
|
|
db_task = crud_task.get_task(db=db, task_id=task_id)
|
|
if db_task is None:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
|
|
@router.put("/{task_id}", response_model=TaskResponse)
|
|
def update_task(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):
|
|
db_task = crud_task.update_task(db=db, task_id=task_id, task_update=task)
|
|
if db_task is None:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
|
|
@router.delete("/{task_id}")
|
|
def delete_task(task_id: int, db: Session = Depends(get_db)):
|
|
success = crud_task.delete_task(db=db, task_id=task_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return {"message": "Task deleted successfully"}
|
|
|
|
|
|
@router.get("/stats/summary")
|
|
def get_task_stats(db: Session = Depends(get_db)):
|
|
total_tasks = crud_task.get_tasks_count(db=db)
|
|
pending_tasks = crud_task.get_tasks_by_status_count(db=db, status=TaskStatus.PENDING)
|
|
in_progress_tasks = crud_task.get_tasks_by_status_count(db=db, status=TaskStatus.IN_PROGRESS)
|
|
completed_tasks = crud_task.get_tasks_by_status_count(db=db, status=TaskStatus.COMPLETED)
|
|
|
|
return {
|
|
"total_tasks": total_tasks,
|
|
"pending_tasks": pending_tasks,
|
|
"in_progress_tasks": in_progress_tasks,
|
|
"completed_tasks": completed_tasks
|
|
} |