from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Literal
from uuid import UUID, uuid4
__all__ = (
"TERMINAL_STATUSES",
"HeartbeatTouch",
"HeartbeatTouchResult",
"QueueBackendCapabilities",
"QueueStatistics",
"QueuedTaskRecord",
"StaleTaskRecoveryResult",
"TaskRequest",
"TaskReservation",
"TaskStatus",
)
TaskStatus = Literal["pending", "scheduled", "running", "completed", "failed", "cancelled", "expired"]
"""Queue task lifecycle states."""
TERMINAL_STATUSES: "frozenset[TaskStatus]" = frozenset({"completed", "failed", "cancelled", "expired"})
"""Statuses that represent finished queue records."""
[docs]
@dataclass(frozen=True, slots=True)
class HeartbeatTouch:
"""A fenced heartbeat update request for one running task."""
task_id: "UUID"
expected_retry_count: "int | None"
metadata_patch: "dict[str, Any] | None" = None
[docs]
@dataclass(slots=True)
class HeartbeatTouchResult:
"""Backend-neutral result for a bulk heartbeat update."""
touched_task_ids: "set[UUID]" = field(default_factory=set)
missed_task_ids: "set[UUID]" = field(default_factory=set)
[docs]
@dataclass(slots=True)
class QueueBackendCapabilities:
"""Behavior advertised by a queue backend."""
supports_worker_wakeups: "bool" = False
wakeup_backend: "str | None" = None
wakeups_durable: "bool" = False
supports_completion_events: "bool" = False
supports_maintenance: "bool" = False
[docs]
@dataclass(slots=True)
class QueueStatistics:
"""Operational status counts for a queue backend."""
pending: "int" = 0
scheduled: "int" = 0
running: "int" = 0
completed: "int" = 0
failed: "int" = 0
cancelled: "int" = 0
expired: "int" = 0
@property
def total(self) -> "int":
"""Total number of known queue records."""
return (
self.pending + self.scheduled + self.running + self.completed + self.failed + self.cancelled + self.expired
)
[docs]
@dataclass(slots=True)
class StaleTaskRecoveryResult:
"""Summary of stale running task recovery."""
requeued: "int" = 0
failed: "int" = 0
skipped: "int" = 0
handler_needed: "int" = 0
failed_task_ids: "list[UUID]" = field(default_factory=list)
handler_needed_task_ids: "list[UUID]" = field(default_factory=list)
[docs]
def to_payload(self) -> "dict[str, int]":
"""Return a JSON-compatible event payload."""
return {
"requeued": self.requeued,
"failed": self.failed,
"skipped": self.skipped,
"handler_needed": self.handler_needed,
}
[docs]
@dataclass(frozen=True, slots=True)
class TaskReservation:
"""A durable ``unique_until="forever"`` identity reservation.
Records only the identity key, the originating task id/name, and the
reservation time. It never carries args, kwargs, result, or error material.
Routine terminal and event maintenance never delete it; an explicit
administrative reset is the only removal path.
"""
key: "str"
task_id: "UUID"
task_name: "str"
created_at: "datetime"
[docs]
@dataclass(slots=True)
class TaskRequest:
"""A task submission for bulk enqueue via ``enqueue_many``.
Carries the task name and enqueue arguments needed to submit one item in a
batch. The backend returns the resulting queue records in request order.
"""
task_name: "str"
args: "tuple[Any, ...]" = ()
kwargs: "dict[str, Any] | None" = None
queue: "str" = "default"
priority: "int" = 0
max_retries: "int" = 0
scheduled_at: "datetime | None" = None
expires_at: "datetime | None" = None
key: "str | None" = None
execution_backend: "str" = "local"
execution_profile: "str | None" = None
metadata: "dict[str, Any] | None" = None
def __post_init__(self) -> "None":
if self.scheduled_at is not None:
self.scheduled_at = _ensure_utc_datetime(self.scheduled_at)
if self.expires_at is not None:
self.expires_at = _ensure_utc_datetime(self.expires_at)
[docs]
@dataclass(slots=True)
class QueuedTaskRecord:
"""Backend-neutral representation of a queued task."""
task_name: "str"
id: "UUID" = field(default_factory=uuid4)
args: "tuple[Any, ...]" = ()
kwargs: "dict[str, Any]" = field(default_factory=dict)
queue: "str" = "default"
execution_backend: "str" = "local"
execution_profile: "str | None" = None
execution_ref: "str | None" = None
worker_id: "str | None" = None
status: "TaskStatus" = "pending"
priority: "int" = 0
max_retries: "int" = 0
retry_count: "int" = 0
scheduled_at: "datetime | None" = None
expires_at: "datetime | None" = None
created_at: "datetime" = field(default_factory=lambda: datetime.now(timezone.utc))
queued_at: "datetime" = field(default_factory=lambda: datetime.now(timezone.utc))
started_at: "datetime | None" = None
completed_at: "datetime | None" = None
heartbeat_at: "datetime | None" = None
result: "Any" = None
error: "str | None" = None
key: "str | None" = None
metadata: "dict[str, Any]" = field(default_factory=dict)
def __post_init__(self) -> "None":
self.created_at = _ensure_utc_datetime(self.created_at)
self.queued_at = _ensure_utc_datetime(self.queued_at)
if self.scheduled_at is not None:
self.scheduled_at = _ensure_utc_datetime(self.scheduled_at)
if self.expires_at is not None:
self.expires_at = _ensure_utc_datetime(self.expires_at)
@property
def is_terminal(self) -> "bool":
"""Whether the record is in a terminal state."""
return self.status in TERMINAL_STATUSES
@property
def is_due(self) -> "bool":
"""Whether the record is eligible to be claimed now."""
return self.scheduled_at is None or self.scheduled_at <= datetime.now(timezone.utc)
@property
def is_expired(self) -> "bool":
"""Whether the not-started deadline has passed."""
return self.expires_at is not None and self.expires_at <= datetime.now(timezone.utc)
def _ensure_utc_datetime(value: "datetime") -> "datetime":
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
