taskmanagerapi-r45vn7/app/api/v1/endpoints/tasks.py

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

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

from app.core.deps import get_current_user, get_db
from app.models.task import Task, TaskPriority, TaskStatus
from app.models.user import User
from app.schemas.task import (
    Task as TaskSchema,
)
from app.schemas.task import (
    TaskCreate,
    TaskUpdate,
    TaskWithAssignees,
)

router = APIRouter()


@router.get("/", response_model=List[TaskSchema])
def read_tasks(
    *,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
    skip: int = 0,
    limit: int = 100,
    status: Optional[TaskStatus] = None,
    priority: Optional[TaskPriority] = None,
    search: Optional[str] = None,
) -> Any:
    """
    Retrieve tasks owned by current user or assigned to current user.
    Optionally filter by status, priority, or search term.
    """
    query = db.query(Task).filter(
        and_(
            or_(
                Task.owner_id == current_user.id,
                Task.assignees.any(id=current_user.id)
            ),
            not Task.is_deleted
        )
    )

    # Apply filters if provided
    if status:
        query = query.filter(Task.status == status)
    if priority:
        query = query.filter(Task.priority == priority)
    if search:
        query = query.filter(
            or_(
                Task.title.ilike(f"%{search}%"),
                Task.description.ilike(f"%{search}%"),
            )
        )

    # Apply pagination
    tasks = query.order_by(Task.updated_at.desc()).offset(skip).limit(limit).all()

    return tasks


@router.post("/", response_model=TaskSchema)
def create_task(
    *,
    db: Session = Depends(get_db),
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
) -> Any:
    """
    Create a new task.
    """
    # Create task object
    task = Task(
        **task_in.model_dump(),
        owner_id=current_user.id,
    )

    # Save to database
    db.add(task)
    db.commit()
    db.refresh(task)

    return task


@router.get("/{task_id}", response_model=TaskWithAssignees)
def read_task(
    *,
    db: Session = Depends(get_db),
    task_id: int,
    current_user: User = Depends(get_current_user),
) -> Any:
    """
    Get details of a specific task.
    """
    task = db.query(Task).filter(
        and_(
            Task.id == task_id,
            not Task.is_deleted
        )
    ).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    # Check if user has access to task
    if task.owner_id != current_user.id and current_user not in task.assignees:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You don't have access to this task"
        )

    # Convert to TaskWithAssignees schema
    result = TaskWithAssignees.model_validate(task)
    result.assignee_ids = [user.id for user in task.assignees]

    return result


@router.put("/{task_id}", response_model=TaskSchema)
def update_task(
    *,
    db: Session = Depends(get_db),
    task_id: int,
    task_in: TaskUpdate,
    current_user: User = Depends(get_current_user),
) -> Any:
    """
    Update a task.
    """
    task = db.query(Task).filter(
        and_(
            Task.id == task_id,
            not Task.is_deleted
        )
    ).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    # Only task owner can update a task
    if task.owner_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only the task owner can update the task"
        )

    # Update with new data
    update_data = task_in.model_dump(exclude_unset=True)

    # If status is being changed to completed, set completed_at
    if "status" in update_data and update_data["status"] == TaskStatus.COMPLETED:
        update_data["completed_at"] = datetime.utcnow()
    # If status is being changed from completed to another status, clear completed_at
    elif "status" in update_data and task.status == TaskStatus.COMPLETED:
        update_data["completed_at"] = None

    # Update task object
    for field, value in update_data.items():
        setattr(task, field, value)

    # Save to database
    db.add(task)
    db.commit()
    db.refresh(task)

    return task


@router.delete(
    "/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    response_model=None
)
def delete_task(
    *,
    db: Session = Depends(get_db),
    task_id: int,
    current_user: User = Depends(get_current_user),
) -> Any:
    """
    Delete a task (soft delete).
    """
    task = db.query(Task).filter(
        and_(
            Task.id == task_id,
            not Task.is_deleted
        )
    ).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    # Only task owner can delete a task
    if task.owner_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only the task owner can delete the task"
        )

    # Soft delete
    task.is_deleted = True

    # Save to database
    db.add(task)
    db.commit()

    return None