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

pythonProject2/routers/v2/tasks.py
150 строк5 KB
Pol136
add pagination and fields
30 дек 2025, 12:09
30 дек 2025, 12:09ee91189
from fastapi import APIRouter, Depends, HTTPException, status, Query, Header
from sqlalchemy.orm import Session
from typing import List, Optional
from fastapi.responses import JSONResponse
from pagination import PaginatedResponse

from database import get_db
from database import Task, User
from schemas import TaskCreate, TaskUpdate, TaskResponse
from security import get_current_user
from fields import parse_fields, filter_object_fields, filter_list_fields, TASK_ALLOWED_FIELDS
from idempotency import (
    get_idempotency_key,
    get_cached_response,
    cache_response
)

router = APIRouter()


@router.post("/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
        task: TaskCreate,
        db: Session = Depends(get_db),
        current_user: User = Depends(get_current_user),
        idempotency_key: Optional[str] = Depends(get_idempotency_key)
):

    if idempotency_key:
        cached = get_cached_response(idempotency_key)
        if cached:
            response, status_code = cached
            return JSONResponse(
                status_code=status_code,
                content=response,
                headers={"Idempotency-Replay": "true", "API-Version": "2.0"}
            )

    db_task = Task(
        owner_id=current_user.id,
        title=task.title,
        description=task.description,
        priority=task.priority
    )
    db.add(db_task)
    db.commit()
    db.refresh(db_task)

    if idempotency_key:
        response_data = {
            "id": db_task.id,
            "owner_id": db_task.owner_id,
            "title": db_task.title,
            "description": db_task.description,
            "status": db_task.status,
            "priority": db_task.priority,
            "created_at": db_task.created_at.isoformat(),
            "updated_at": db_task.updated_at.isoformat(),
            "completed_at": db_task.completed_at,
            "api_version": "2.0"
        }
        cache_response(idempotency_key, response_data, 201)

    return db_task


@router.get("/", response_model=dict)
async def list_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 to include (e.g., id,title,status)"),
        db: Session = Depends(get_db),
        current_user: User = Depends(get_current_user)
):
    query = db.query(Task).filter(Task.owner_id == current_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("/{task_id}", response_model=dict)
async def get_task(
        task_id: int,
        fields: Optional[str] = Query(None, description="Comma-separated field names to include"),
        db: Session = Depends(get_db),
        current_user: User = Depends(get_current_user)
):
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()

    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    requested_fields = parse_fields(fields, TASK_ALLOWED_FIELDS)
    return filter_object_fields(task, requested_fields)


@router.patch("/{task_id}", response_model=TaskResponse)
async def update_task(
        task_id: int,
        task_update: TaskUpdate,
        db: Session = Depends(get_db),
        current_user: User = Depends(get_current_user)
):
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    update_data = task_update.dict(exclude_unset=True)
    for field, value in update_data.items():
        setattr(task, field, value)

    db.add(task)
    db.commit()
    db.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
        task_id: int,
        db: Session = Depends(get_db),
        current_user: User = Depends(get_current_user)
):
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    db.delete(task)
    db.commit()