Source code for edri.abstract.manager.worker

from __future__ import annotations

from dataclasses import dataclass, field
from multiprocessing.connection import Connection
from pathlib import Path
from typing import Any, Protocol
from uuid import UUID

from edri.dataclass.event import Event
from edri.dataclass.worker import (
    WorkerExit, WorkerOutcome, WorkerPolicy, WorkerState, WorkerStopReason,
    WorkerTimeout,
)




class WorkerRuntimeProtocol(Protocol):
    """Runtime operations owned by :class:`WorkerController`."""

    name: str



    def start(self) -> None: ...




    def is_alive(self) -> bool: ...


    def _bind_lifecycle(self, cancel_event: Any) -> None: ...

    def _bind_router_queue(self, router_queue: Any) -> None: ...

    def _request_cancel(self) -> None: ...





class WorkerFactoryProtocol(Protocol):
    """Constructor contract for thread and process worker classes."""

    def __call__(
        self,
        pipe: Connection,
        event: Event | None,
        name: str,
        /,
        *args: Any,
        **kwargs: Any,
    ) -> WorkerRuntimeProtocol: ...





@dataclass(slots=True)
class Worker:
    event: Event | None
    resolver: WorkerFactoryProtocol
    args: tuple[Any, ...] = ()
    kwargs: dict[str, Any] = field(default_factory=dict)
    environment: Path | None = None
    policy: WorkerPolicy = field(default_factory=WorkerPolicy)
    state: WorkerState = WorkerState.PENDING
    pipe: Connection | None = None
    runtime: WorkerRuntimeProtocol | None = None
    created: float = 0.0
    spawned: float | None = None
    started: float | None = None
    stop_requested: float | None = None
    finished: float | None = None
    stop_reason: WorkerStopReason | None = None
    timeout: WorkerTimeout | None = None
    outcome: WorkerOutcome | None = None
    exit: WorkerExit | None = None
    pending_responses: set[UUID] = field(default_factory=set)





class WorkerNotFoundError(KeyError):
    """Raised when cancellation targets a worker that is no longer registered."""



__all__ = [
    "Worker", "WorkerExit", "WorkerFactoryProtocol", "WorkerNotFoundError",
    "WorkerOutcome", "WorkerPolicy", "WorkerRuntimeProtocol", "WorkerState",
    "WorkerStopReason", "WorkerTimeout",
]