72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
from typing import Optional, Dict, Any
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import Task
|
|
from app.schemas import TaskCreate, TaskUpdate, TaskResponse, TaskList
|
|
from app.crud import create_task, get_task, get_tasks, get_tasks_count, update_task, delete_task
|
|
|
|
router = APIRouter(tags=["tasks"])
|
|
|
|
@router.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_task_endpoint(task: TaskCreate, db: Session = Depends(get_db)):
|
|
"""
|
|
Create a new task
|
|
"""
|
|
return create_task(db=db, task=task)
|
|
|
|
@router.get("/tasks", response_model=TaskList)
|
|
def get_tasks_endpoint(
|
|
status: Optional[str] = Query(None, description="Filter by status"),
|
|
priority: Optional[str] = Query(None, description="Filter by priority"),
|
|
completed: Optional[bool] = Query(None, description="Filter by completion status"),
|
|
skip: int = Query(0, ge=0, description="Number of tasks to skip"),
|
|
limit: int = Query(100, ge=1, le=100, description="Max number of tasks to return"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Get list of tasks with optional filtering
|
|
"""
|
|
filters: Dict[str, Any] = {}
|
|
if status:
|
|
filters["status"] = status
|
|
if priority:
|
|
filters["priority"] = priority
|
|
if completed is not None:
|
|
filters["completed"] = completed
|
|
|
|
tasks = get_tasks(db=db, skip=skip, limit=limit, filters=filters)
|
|
total = get_tasks_count(db=db, filters=filters)
|
|
|
|
return {"tasks": tasks, "total": total}
|
|
|
|
@router.get("/tasks/{task_id}", response_model=TaskResponse)
|
|
def get_task_endpoint(task_id: int, db: Session = Depends(get_db)):
|
|
"""
|
|
Get a specific task by ID
|
|
"""
|
|
db_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("/tasks/{task_id}", response_model=TaskResponse)
|
|
def update_task_endpoint(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):
|
|
"""
|
|
Update a task
|
|
"""
|
|
db_task = update_task(db=db, task_id=task_id, task=task)
|
|
if db_task is None:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return db_task
|
|
|
|
@router.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_task_endpoint(task_id: int, db: Session = Depends(get_db)):
|
|
"""
|
|
Delete a task
|
|
"""
|
|
success = delete_task(db=db, task_id=task_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
return None |