Skip to content

pop_strategy

JobPopStrategyGeneric

Bases: ABC

Abstract base class for job pop strategies.

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class JobPopStrategyGeneric[SingleGenerationTypeVar, ComposedParameterSetTypeVar](ABC):
    """Abstract base class for job pop strategies."""

    _default_job_pop_time_spacing: float = 1.0
    """Default minimum time spacing between job pops in seconds."""

    def __init__(
        self,
        default_job_pop_time_spacing: float = _default_job_pop_time_spacing,
    ) -> None:
        """Initialize the job pop strategy."""
        self._default_job_pop_time_spacing = default_job_pop_time_spacing

    @abstractmethod
    def get_worker_type(self) -> WORKER_TYPE:
        """Get the worker type associated with this job pop strategy.

        Returns:
            WORKER_TYPE: The worker type.
        """

    @abstractmethod
    def get_dispatch_source(self) -> KNOWN_DISPATCH_SOURCE | str:
        """Get the dispatch source associated with this job pop strategy.

        Returns:
            KNOWN_DISPATCH_SOURCE | str: The dispatch source.
        """

    @abstractmethod
    def pop_job(self) -> HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None:
        """Pop a job synchronously from the dispatch source.

        Use `async_pop_job` for asynchronous operations.

        Returns:
            HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None: The popped job or `None` if
            no job is available.
        """

    @abstractmethod
    async def async_pop_job(self) -> HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None:
        """Pop a job asynchronously from the dispatch source.

        Use `pop_job` if you prefer synchronous operations.

        Returns:
            HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None: The popped job or `None` if
            no job is available.
        """

__init__ 

__init__(
    default_job_pop_time_spacing: float = _default_job_pop_time_spacing,
) -> None

Initialize the job pop strategy.

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
17
18
19
20
21
22
def __init__(
    self,
    default_job_pop_time_spacing: float = _default_job_pop_time_spacing,
) -> None:
    """Initialize the job pop strategy."""
    self._default_job_pop_time_spacing = default_job_pop_time_spacing

get_worker_type abstractmethod 

get_worker_type() -> WORKER_TYPE

Get the worker type associated with this job pop strategy.

Returns:

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
24
25
26
27
28
29
30
@abstractmethod
def get_worker_type(self) -> WORKER_TYPE:
    """Get the worker type associated with this job pop strategy.

    Returns:
        WORKER_TYPE: The worker type.
    """

get_dispatch_source abstractmethod 

get_dispatch_source() -> KNOWN_DISPATCH_SOURCE | str

Get the dispatch source associated with this job pop strategy.

Returns:

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
32
33
34
35
36
37
38
@abstractmethod
def get_dispatch_source(self) -> KNOWN_DISPATCH_SOURCE | str:
    """Get the dispatch source associated with this job pop strategy.

    Returns:
        KNOWN_DISPATCH_SOURCE | str: The dispatch source.
    """

pop_job abstractmethod 

pop_job() -> (
    HordeWorkerJob[
        HordeSingleGeneration[Any], CompositeParametersBase
    ]
    | None
)

Pop a job synchronously from the dispatch source.

Use async_pop_job for asynchronous operations.

Returns:

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
40
41
42
43
44
45
46
47
48
49
@abstractmethod
def pop_job(self) -> HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None:
    """Pop a job synchronously from the dispatch source.

    Use `async_pop_job` for asynchronous operations.

    Returns:
        HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None: The popped job or `None` if
        no job is available.
    """

async_pop_job abstractmethod async 

async_pop_job() -> (
    HordeWorkerJob[
        HordeSingleGeneration[Any], CompositeParametersBase
    ]
    | None
)

Pop a job asynchronously from the dispatch source.

Use pop_job if you prefer synchronous operations.

Returns:

Source code in horde_sdk/worker/dispatch/pop_strategy.py 
51
52
53
54
55
56
57
58
59
60
@abstractmethod
async def async_pop_job(self) -> HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None:
    """Pop a job asynchronously from the dispatch source.

    Use `pop_job` if you prefer synchronous operations.

    Returns:
        HordeWorkerJob[HordeSingleGeneration[Any], CompositeParametersBase] | None: The popped job or `None` if
        no job is available.
    """