mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-08-02 21:08:24 +00:00
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from enum import Enum
|
|
from typing import Optional
|
|
from uuid import UUID, uuid4
|
|
|
|
from domain.tasks.subtasks import SubTaskStatus
|
|
|
|
|
|
class TaskStatus(str, Enum):
|
|
WAITING = "waiting"
|
|
IN_PROGRESS = "in progress"
|
|
COMPLETE = "complete"
|
|
FAILED = "failed"
|
|
|
|
|
|
class Source(str, Enum):
|
|
PORTFOLIO = "portfolio_id"
|
|
HUBSPOT_DEAL = "hubspot_deal_id"
|
|
PROPERTY = "property_id"
|
|
# A run anchored on a SharePoint site rather than anything in our own
|
|
# schema — the SharePoint Renamer. The pgEnum is FE-owned via Drizzle
|
|
# (ADR-0003), so the Drizzle migration adding 'sharepoint_site' MUST apply
|
|
# before this deploys or the first run fails on an invalid enum value.
|
|
# Tests won't catch that: ephemeral DBs build from the SQLModel mirrors
|
|
# below and emit the member automatically.
|
|
SHAREPOINT_SITE = "sharepoint_site"
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
id: UUID
|
|
task_source: str
|
|
status: TaskStatus = TaskStatus.WAITING
|
|
service: Optional[str] = None
|
|
source: Optional[Source] = None
|
|
source_id: Optional[str] = None
|
|
job_started: Optional[datetime] = None
|
|
job_completed: Optional[datetime] = None
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
*,
|
|
task_source: str,
|
|
service: Optional[str] = None,
|
|
source: Optional[Source] = None,
|
|
source_id: Optional[str] = None,
|
|
) -> "Task":
|
|
if not task_source.strip():
|
|
raise ValueError("task_source must be non-empty")
|
|
return cls(
|
|
id=uuid4(),
|
|
task_source=task_source,
|
|
service=service,
|
|
source=source,
|
|
source_id=source_id,
|
|
status=TaskStatus.WAITING,
|
|
job_started=datetime.now(timezone.utc),
|
|
)
|
|
|
|
def start(self) -> None:
|
|
if self.status not in (TaskStatus.WAITING, TaskStatus.IN_PROGRESS):
|
|
raise ValueError(f"cannot start task in status {self.status}")
|
|
if self.job_started is None:
|
|
self.job_started = datetime.now(timezone.utc)
|
|
self.status = TaskStatus.IN_PROGRESS
|
|
|
|
def complete(self) -> None:
|
|
self.status = TaskStatus.COMPLETE
|
|
self.job_completed = datetime.now(timezone.utc)
|
|
|
|
def fail(self) -> None:
|
|
self.status = TaskStatus.FAILED
|
|
self.job_completed = datetime.now(timezone.utc)
|
|
|
|
def recalculate_from_subtasks(self, statuses: list[SubTaskStatus]) -> None:
|
|
"""Recompute Task.status from its SubTasks' statuses.
|
|
|
|
Rule:
|
|
- any FAILED → FAILED
|
|
- all COMPLETE → COMPLETE
|
|
- any IN_PROGRESS or COMPLETE → IN_PROGRESS (finished batches plus
|
|
queued batches is a run in progress, not a waiting one — ADR-0055)
|
|
- all WAITING → WAITING
|
|
|
|
Empty list is a no-op (newly-created task with no subtasks).
|
|
"""
|
|
if not statuses:
|
|
return
|
|
now = datetime.now(timezone.utc)
|
|
if SubTaskStatus.FAILED in statuses:
|
|
self.status = TaskStatus.FAILED
|
|
self.job_completed = now
|
|
elif all(s is SubTaskStatus.COMPLETE for s in statuses):
|
|
self.status = TaskStatus.COMPLETE
|
|
self.job_completed = now
|
|
elif (
|
|
SubTaskStatus.IN_PROGRESS in statuses
|
|
or SubTaskStatus.COMPLETE in statuses
|
|
):
|
|
self.status = TaskStatus.IN_PROGRESS
|
|
self.job_completed = None
|
|
else:
|
|
self.status = TaskStatus.WAITING
|
|
self.job_completed = None
|