services/task_service.py

from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import List, Optional
from models.task import Task, TaskBase
from schemas.task import TaskCreate, TaskUpdate
import uuid

async def create_task(*, session: AsyncSession, task_in: TaskCreate, user_id: uuid.UUID) -> Task:
    """
    Create a new task for the given user
    """
    task = Task.from_orm(task_in)
    task.user_id = user_id
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

async def get_tasks_by_user(*, session: AsyncSession, user_id: uuid.UUID) -> List[Task]:
    """
    Get all tasks for a specific user
    """
    statement = select(Task).where(Task.user_id == user_id)
    result = await session.exec(statement)
    tasks = result.all()
    return tasks

async def get_task_by_id(*, session: AsyncSession, task_id: uuid.UUID, user_id: uuid.UUID) -> Optional[Task]:
    """
    Get a specific task by ID for a specific user
    """
    statement = select(Task).where(Task.id == task_id, Task.user_id == user_id)
    result = await session.exec(statement)
    task = result.first()
    return task

async def update_task(*, session: AsyncSession, task: Task, task_in: TaskUpdate) -> Task:
    """
    Update a task with the provided data
    """
    task_data = task_in.dict(exclude_unset=True)
    for field, value in task_data.items():
        setattr(task, field, value)

    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

async def delete_task(*, session: AsyncSession, task: Task) -> bool:
    """
    Delete a task
    """
    await session.delete(task)
    await session.commit()
    return True

async def update_task_completion_status(*, session: AsyncSession, task: Task, completed: bool) -> Task:
    """
    Update the completion status of a task
    """
    task.completed = completed
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task