pythonProject2/routers/internal/tasks.py - feature/lab2 - Pepegator/task-api

pythonProject2/routers/internal/tasks.py
88 строк3 KB
Pol136
add internal api
30 дек 2025, 12:39
30 дек 2025, 12:39b881890
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import Optional, Dict, Any

from database import get_db, User, Task
from pagination import PaginatedResponse
from fields import parse_fields, filter_list_fields, TASK_ALLOWED_FIELDS

router = APIRouter()


@router.get("/all", response_model=dict)
async def get_all_tasks(
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1, le=100),
        status: Optional[str] = None,
        fields: Optional[str] = Query(None, description="Comma-separated field names"),
        db: Session = Depends(get_db)
):
    query = db.query(Task)

    if status:
        query = query.filter(Task.status == status)

    total = query.count()
    tasks = query.offset(skip).limit(limit).all()

    requested_fields = parse_fields(fields, TASK_ALLOWED_FIELDS)
    filtered_tasks = filter_list_fields(tasks, requested_fields)

    response = PaginatedResponse.create(tasks, total, skip, limit).model_dump()
    response["items"] = filtered_tasks
    return response


@router.get("/user/{user_id}", response_model=dict)
async def get_user_tasks(
        user_id: int,
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1, le=100),
        status: Optional[str] = None,
        fields: Optional[str] = Query(None, description="Comma-separated field names"),
        db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    query = db.query(Task).filter(Task.owner_id == user_id)

    if status:
        query = query.filter(Task.status == status)

    total = query.count()
    tasks = query.offset(skip).limit(limit).all()

    requested_fields = parse_fields(fields, TASK_ALLOWED_FIELDS)
    filtered_tasks = filter_list_fields(tasks, requested_fields)

    response = PaginatedResponse.create(tasks, total, skip, limit).model_dump()
    response["items"] = filtered_tasks
    return response


@router.get("/user/{user_id}/count", response_model=Dict[str, Any])
async def get_user_task_count(
        user_id: int,
        db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    total = db.query(Task).filter(Task.owner_id == user_id).count()
    todo = db.query(Task).filter(Task.owner_id == user_id, Task.status == "todo").count()
    in_progress = db.query(Task).filter(Task.owner_id == user_id, Task.status == "in_progress").count()
    done = db.query(Task).filter(Task.owner_id == user_id, Task.status == "done").count()

    return {
        "user_id": user_id,
        "username": user.username,
        "total_tasks": total,
        "tasks_by_status": {
            "todo": todo,
            "in_progress": in_progress,
            "done": done
        }
    }