task_management_api/app/services/task_service.py - master - tinypot/integr

task_management_api/app/services/task_service.py
100 строк4 KB
Tim Polus
lab1
21 янв 2026, 23:15
21 янв 2026, 23:15dd1578c
"""
Task service for business logic
"""

from typing import Optional, List
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.models.task import Task, TaskStatus
from app.schemas.task import TaskCreate, TaskUpdate


class TaskService:
    """Service class for task operations"""

    @staticmethod
    async def get_by_id(db: AsyncSession, task_id: int) -> Optional[Task]:
        """Get task by ID"""
        stmt = select(Task).where(Task.id == task_id)
        result = await db.execute(stmt)
        return result.scalar_one_or_none()

    @staticmethod
    async def get_by_user_id(db: AsyncSession, user_id: int, skip: int = 0, limit: int = 100) -> List[Task]:
        """Get tasks by user ID"""
        stmt = select(Task).where(Task.user_id == user_id).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def get_by_category_id(db: AsyncSession, category_id: int, user_id: int, skip: int = 0, limit: int = 100) -> List[Task]:
        """Get tasks by category ID and user ID"""
        stmt = select(Task).where(Task.category_id == category_id, Task.user_id == user_id).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def get_by_status(db: AsyncSession, user_id: int, status: TaskStatus, skip: int = 0, limit: int = 100) -> List[Task]:
        """Get tasks by status and user ID"""
        stmt = select(Task).where(Task.user_id == user_id, Task.status == status).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def get_all(db: AsyncSession, skip: int = 0, limit: int = 100) -> List[Task]:
        """Get all tasks with pagination"""
        stmt = select(Task).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def create(db: AsyncSession, task_in: TaskCreate, user_id: int) -> Task:
        """Create new task"""
        db_task = Task(
            title=task_in.title,
            description=task_in.description,
            status=task_in.status,
            priority=task_in.priority,
            due_date=task_in.due_date,
            category_id=task_in.category_id,
            user_id=user_id,
        )

        db.add(db_task)
        await db.commit()
        await db.refresh(db_task)
        return db_task

    @staticmethod
    async def update(db: AsyncSession, db_task: Task, task_in: TaskUpdate) -> Task:
        """Update task"""
        update_data = task_in.dict(exclude_unset=True)

        # Set completed_at when status changes to completed
        if "status" in update_data and update_data["status"] == TaskStatus.COMPLETED:
            update_data["completed_at"] = datetime.utcnow()
        elif "status" in update_data and update_data["status"] != TaskStatus.COMPLETED:
            update_data["completed_at"] = None

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

        await db.commit()
        await db.refresh(db_task)
        return db_task

    @staticmethod
    async def delete(db: AsyncSession, db_task: Task) -> None:
        """Delete task"""
        await db.delete(db_task)
        await db.commit()

    @staticmethod
    async def get_user_task(db: AsyncSession, task_id: int, user_id: int) -> Optional[Task]:
        """Get task by ID and ensure it belongs to user"""
        stmt = select(Task).where(Task.id == task_id, Task.user_id == user_id)
        result = await db.execute(stmt)
        return result.scalar_one_or_none()