Cheng Zhou
78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
task_in: TaskCreate,
|
|
*,
|
|
created_by: uuid.UUID,
|
|
) -> Task:
|
|
task = Task(
|
|
study_id=study_id,
|
|
milestone_id=task_in.milestone_id,
|
|
title=task_in.title,
|
|
description=task_in.description,
|
|
assignee_id=task_in.assignee_id,
|
|
priority=task_in.priority,
|
|
due_date=task_in.due_date,
|
|
status="TODO",
|
|
created_by=created_by,
|
|
)
|
|
db.add(task)
|
|
await db.commit()
|
|
await db.refresh(task)
|
|
return task
|
|
|
|
|
|
async def get(db: AsyncSession, task_id: uuid.UUID) -> Task | None:
|
|
result = await db.execute(select(Task).where(Task.id == task_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_tasks(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
milestone_id: uuid.UUID | None = None,
|
|
assignee_id: uuid.UUID | None = None,
|
|
status: str | None = None,
|
|
) -> Sequence[Task]:
|
|
stmt = select(Task).where(Task.study_id == study_id)
|
|
if milestone_id:
|
|
stmt = stmt.where(Task.milestone_id == milestone_id)
|
|
if assignee_id:
|
|
stmt = stmt.where(Task.assignee_id == assignee_id)
|
|
if status:
|
|
stmt = stmt.where(Task.status == status)
|
|
stmt = stmt.order_by(Task.due_date)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update(db: AsyncSession, task: Task, task_in: TaskUpdate) -> Task:
|
|
update_data = task_in.model_dump(exclude_unset=True)
|
|
if "status" in update_data:
|
|
status_change = update_data["status"]
|
|
if status_change == "DONE":
|
|
update_data["completed_at"] = datetime.now(timezone.utc)
|
|
else:
|
|
update_data["completed_at"] = None
|
|
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(Task)
|
|
.where(Task.id == task.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(task)
|
|
return task
|