task_management_api/app/api/v1/tasks.py - master - tinypot/integr

task_management_api/app/api/v1/tasks.py
149 строк4 KB
Tim Polus
lab1
21 янв 2026, 23:15
21 янв 2026, 23:15dd1578c
"""
Tasks API endpoints
"""

from typing import Any, List

from fastapi import APIRouter, Depends, HTTPException, status, Request
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.models.user import User
from app.schemas.task import Task, TaskCreate, TaskUpdate, TaskStatus
from app.services.task_service import TaskService
from app.utils.idempotency import idempotency_middleware

router = APIRouter()


@router.get("/", response_model=List[Task])
async def read_tasks(
    skip: int = 0,
    limit: int = 100,
    status_filter: TaskStatus = None,
    category_id: int = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """
    Retrieve tasks for current user with optional filters
    """
    if status_filter:
        tasks = await TaskService.get_by_status(
            db, user_id=current_user.id, status=status_filter, skip=skip, limit=limit
        )
    elif category_id:
        tasks = await TaskService.get_by_category_id(
            db, category_id=category_id, user_id=current_user.id, skip=skip, limit=limit
        )
    else:
        tasks = await TaskService.get_by_user_id(
            db, user_id=current_user.id, skip=skip, limit=limit
        )
    return tasks


@router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task(
    request: Request,
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """
    Create new task with idempotency support
    """
    # Apply idempotency middleware
    response = await idempotency_middleware(request, lambda: None, current_user.id)

    # Check if category belongs to user (if provided)
    if task_in.category_id:
        from app.services.category_service import CategoryService
        category = await CategoryService.get_user_category(
            db, category_id=task_in.category_id, user_id=current_user.id
        )
        if not category:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Category not found or does not belong to user"
            )

    task = await TaskService.create(db, task_in=task_in, user_id=current_user.id)
    return task


@router.get("/{task_id}", response_model=Task)
async def read_task(
    task_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """
    Get task by ID
    """
    task = await TaskService.get_user_task(
        db, task_id=task_id, user_id=current_user.id
    )
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )
    return task


@router.put("/{task_id}", response_model=Task)
async def update_task(
    task_id: int,
    task_in: TaskUpdate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """
    Update task
    """
    task = await TaskService.get_user_task(
        db, task_id=task_id, user_id=current_user.id
    )
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    # Check if category belongs to user (if provided)
    if task_in.category_id:
        from app.services.category_service import CategoryService
        category = await CategoryService.get_user_category(
            db, category_id=task_in.category_id, user_id=current_user.id
        )
        if not category:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Category not found or does not belong to user"
            )

    task = await TaskService.update(db, db_task=task, task_in=task_in)
    return task


@router.delete("/{task_id}")
async def delete_task(
    task_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """
    Delete task
    """
    task = await TaskService.get_user_task(
        db, task_id=task_id, user_id=current_user.id
    )
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )
    await TaskService.delete(db, db_task=task)
    return {"message": "Task deleted successfully"}