Source code for edri.dataclass.worker

from dataclasses import dataclass
from enum import Enum, auto
from math import isfinite




class WorkerState(Enum):
    PENDING = auto()
    STARTING = auto()
    RUNNING = auto()
    STOPPING = auto()
    TERMINATING = auto()





class WorkerOutcome(Enum):
    SUCCEEDED = auto()
    FAILED = auto()
    CANCELLED = auto()
    TIMED_OUT = auto()





class WorkerExit(Enum):
    RETURNED = auto()
    RAISED = auto()
    CANCELLED = auto()
    TERMINATED = auto()
    KILLED = auto()
    LOST = auto()





class WorkerStopReason(Enum):
    REQUESTED = auto()
    TIMEOUT = auto()
    MANAGER_SHUTDOWN = auto()
    CLIENT_DISCONNECT = auto()
    APPLICATION = auto()





class WorkerTimeout(Enum):
    QUEUE = auto()
    START = auto()
    RUN = auto()
    TOTAL = auto()





@dataclass(frozen=True, slots=True)
class WorkerPolicy:
    queue_timeout: float | None = None
    start_timeout: float | None = None
    run_timeout: float | None = None
    total_timeout: float | None = None
    cancel_grace: float = 5.0
    terminate_grace: float = 2.0

    def __post_init__(self) -> None:
        for name in (
            "queue_timeout", "start_timeout", "run_timeout", "total_timeout",
            "cancel_grace", "terminate_grace",
        ):
            value = getattr(self, name)
            if value is not None and (not isfinite(value) or value < 0):
                raise ValueError(f"{name} must be finite and greater than or equal to zero")



__all__ = [
    "WorkerExit", "WorkerOutcome", "WorkerPolicy", "WorkerState",
    "WorkerStopReason", "WorkerTimeout",
]