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

task_management_api/app/api/v2/tasks.py
230 строк7 KB
Tim Polus
lab2
16 фев 2026, 19:22
16 фев 2026, 19:22ec900ad
"""
API задач v2
"""

from typing import Any, List, Optional
from datetime import date, datetime

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, TaskPriority,
    PaginatedTasksResponse, TaskListRequest, TaskPartial
)
from app.services.task_service_v2 import TaskServiceV2
from app.utils.idempotency import idempotency_middleware

router = APIRouter()


@router.get("/", response_model=PaginatedTasksResponse)
async def read_tasks_v2(
    page: Optional[int] = 1,
    per_page: Optional[int] = 10,
    include: Optional[str] = None,
    status_filter: Optional[TaskStatus] = None,
    priority_filter: Optional[TaskPriority] = None,
    category_id: Optional[int] = None,
    due_date_from: Optional[date] = None,
    due_date_to: Optional[date] = None,
    completed_from: Optional[datetime] = None,
    completed_to: Optional[datetime] = None,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    # Parse include fields
    include_fields = None
    if include:
        include_fields = [field.strip() for field in include.split(',') if field.strip()]

    # Get paginated results
    result = await TaskServiceV2.get_paginated_tasks(
        db=db,
        user_id=current_user.id,
        page=page,
        per_page=per_page,
        status=status_filter,
        priority=priority_filter,
        category_id=category_id,
        due_date_from=due_date_from,
        due_date_to=due_date_to,
        completed_from=completed_from,
        completed_to=completed_to,
    )

    # Apply field selection if specified
    if include_fields:
        filtered_items = []
        for task in result["items"]:
            filtered_task = TaskServiceV2.apply_field_selection(task, include_fields)
            filtered_items.append(filtered_task)
        result["items"] = filtered_items

    return result


@router.get("/statistics")
async def get_task_statistics(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    stats = await TaskServiceV2.get_task_statistics(db, user_id=current_user.id)
    return stats


@router.get("/due-soon")
async def get_tasks_due_soon(
    days: int = 7,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    tasks = await TaskServiceV2.get_tasks_due_soon(
        db, user_id=current_user.id, days=days
    )
    return tasks


@router.post("/bulk-update-status")
async def bulk_update_task_status(
    task_ids: List[int],
    new_status: TaskStatus,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    if not task_ids:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Task IDs list cannot be empty"
        )

    updated_count = await TaskServiceV2.bulk_update_status(
        db, user_id=current_user.id, task_ids=task_ids, new_status=new_status
    )

    return {
        "message": f"Updated {updated_count} tasks",
        "updated_count": updated_count
    }


@router.get("/internal/summary", include_in_schema=False)
async def get_internal_task_summary(
    days: int = 30,
    api_key: str = None,
    db: AsyncSession = Depends(get_db),
) -> Any:
    # Simple API key validation for internal access
    # In production, this should be more sophisticated
    expected_key = "internal-api-key-2024"  # Should be in environment variables

    if not api_key or api_key != expected_key:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid or missing API key"
        )

    try:
        summary = await TaskServiceV2.get_internal_task_summary(db, days=days)
        return summary
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Internal error: {str(e)}"
        )


@router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task_v2(
    request: Request,
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    # 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 TaskServiceV2.create(db, task_in=task_in, user_id=current_user.id)
    return task


@router.get("/{task_id}", response_model=Task)
async def read_task_v2(
    task_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    task = await TaskServiceV2.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_v2(
    task_id: int,
    task_in: TaskUpdate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    task = await TaskServiceV2.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 TaskServiceV2.update(db, db_task=task, task_in=task_in)
    return task


@router.delete("/{task_id}")
async def delete_task_v2(
    task_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Any:
    task = await TaskServiceV2.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 TaskServiceV2.delete(db, db_task=task)
    return {"message": "Task deleted successfully"}