pythonProject2/routers/v1/tasks.py - master - Pepegator/task-api

pythonProject2/routers/v1/tasks.py
138 строк4 KB
Pol136
create docs
30 дек 2025, 11:01
30 дек 2025, 11:01d230838
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 database import get_db
from database import Task, User
from schemas import TaskCreate, TaskUpdate, TaskResponse
from security import get_current_user
from idempotency import (
    get_idempotency_key,
    get_cached_response,
    cache_response,
    clear_expired_cache
)

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"}
            )

    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
        }
        cache_response(idempotency_key, response_data, 201)

    return db_task


@router.get("/", response_model=List[TaskResponse])
async def list_tasks(
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1, le=100),
        status_filter: Optional[str] = Query(None, alias="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_filter:
        query = query.filter(Task.status == status_filter)

    return query.offset(skip).limit(limit).all()


@router.get("/{task_id}", response_model=TaskResponse)
async def get_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")

    return task


@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()