158 lines
3.8 KiB
Python
158 lines
3.8 KiB
Python
from datetime import datetime, time, timezone
|
|
|
|
from sqlalchemy import case, or_, select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.extensions import db
|
|
|
|
from .models import TASK_PRIORITIES, TASK_STATUSES, Subtask, Task
|
|
|
|
|
|
class TaskNotFoundError(LookupError):
|
|
pass
|
|
|
|
|
|
def _task_query(user_id: int):
|
|
return (
|
|
select(Task)
|
|
.where(Task.user_id == user_id)
|
|
.options(selectinload(Task.subtasks))
|
|
)
|
|
|
|
|
|
def _task_order():
|
|
priority_order = case(
|
|
{"urgent": 0, "important": 1, "normal": 2},
|
|
value=Task.priority,
|
|
else_=3,
|
|
)
|
|
return (
|
|
case((Task.status == "done", 1), else_=0),
|
|
priority_order,
|
|
Task.due_date.is_(None),
|
|
Task.due_date,
|
|
Task.created_at,
|
|
)
|
|
|
|
|
|
def get_task(task_id: int, user_id: int) -> Task:
|
|
task = db.session.scalar(_task_query(user_id).where(Task.id == task_id))
|
|
if task is None:
|
|
raise TaskNotFoundError
|
|
return task
|
|
|
|
|
|
def list_tasks(user_id: int, status: str | None = None) -> list[Task]:
|
|
query = _task_query(user_id)
|
|
if status is not None:
|
|
if status not in TASK_STATUSES:
|
|
raise ValueError("Unknown task status.")
|
|
query = query.where(Task.status == status)
|
|
return list(db.session.scalars(query.order_by(*_task_order())))
|
|
|
|
|
|
def list_today_tasks(
|
|
user_id: int,
|
|
*,
|
|
now: datetime | None = None,
|
|
limit: int = 7,
|
|
) -> list[Task]:
|
|
local_now = now or datetime.now().astimezone()
|
|
end_of_today = datetime.combine(local_now.date(), time.max)
|
|
|
|
query = (
|
|
_task_query(user_id)
|
|
.where(Task.status != "done")
|
|
.where(or_(Task.due_date.is_(None), Task.due_date <= end_of_today))
|
|
.order_by(*_task_order())
|
|
.limit(limit)
|
|
)
|
|
return list(db.session.scalars(query))
|
|
|
|
|
|
def create_task(
|
|
user_id: int,
|
|
*,
|
|
title: str,
|
|
description: str | None,
|
|
status: str,
|
|
priority: str,
|
|
due_date: datetime | None,
|
|
) -> Task:
|
|
_validate_choices(status, priority)
|
|
task = Task(
|
|
user_id=user_id,
|
|
title=title.strip(),
|
|
description=_clean_optional(description),
|
|
status=status,
|
|
priority=priority,
|
|
due_date=due_date,
|
|
)
|
|
if status == "done":
|
|
task.completed_at = datetime.now(timezone.utc)
|
|
db.session.add(task)
|
|
db.session.commit()
|
|
return task
|
|
|
|
|
|
def update_task(
|
|
task: Task,
|
|
*,
|
|
title: str,
|
|
description: str | None,
|
|
status: str,
|
|
priority: str,
|
|
due_date: datetime | None,
|
|
) -> Task:
|
|
_validate_choices(status, priority)
|
|
was_done = task.status == "done"
|
|
|
|
task.title = title.strip()
|
|
task.description = _clean_optional(description)
|
|
task.status = status
|
|
task.priority = priority
|
|
task.due_date = due_date
|
|
|
|
if status == "done" and not was_done:
|
|
task.completed_at = datetime.now(timezone.utc)
|
|
elif status != "done":
|
|
task.completed_at = None
|
|
|
|
db.session.commit()
|
|
return task
|
|
|
|
|
|
def delete_task(task: Task) -> None:
|
|
db.session.delete(task)
|
|
db.session.commit()
|
|
|
|
|
|
def add_subtask(task: Task, title: str) -> Subtask:
|
|
subtask = Subtask(task=task, title=title.strip())
|
|
db.session.add(subtask)
|
|
db.session.commit()
|
|
return subtask
|
|
|
|
|
|
def toggle_subtask(task: Task, subtask_id: int) -> Subtask:
|
|
subtask = next(
|
|
(item for item in task.subtasks if item.id == subtask_id),
|
|
None,
|
|
)
|
|
if subtask is None:
|
|
raise TaskNotFoundError
|
|
subtask.completed = not subtask.completed
|
|
db.session.commit()
|
|
return subtask
|
|
|
|
|
|
def _validate_choices(status: str, priority: str) -> None:
|
|
if status not in TASK_STATUSES:
|
|
raise ValueError("Unknown task status.")
|
|
if priority not in TASK_PRIORITIES:
|
|
raise ValueError("Unknown task priority.")
|
|
|
|
|
|
def _clean_optional(value: str | None) -> str | None:
|
|
cleaned = value.strip() if value else ""
|
|
return cleaned or None
|