Skip to content

omnipy.shared.protocols.hub.registry

Protocols for tracking job run-state transitions.

CLASS DESCRIPTION
IsRunStateRegistry

Protocol for registries that track job run states.

IsRunStateRegistry

Bases: Protocol


              flowchart BT
              omnipy.shared.protocols.hub.registry.IsRunStateRegistry[IsRunStateRegistry]

              

              click omnipy.shared.protocols.hub.registry.IsRunStateRegistry href "" "omnipy.shared.protocols.hub.registry.IsRunStateRegistry"

Protocol for registries that track job run states.

Returned by: Used by:
METHOD DESCRIPTION
__init__
all_jobs

Return all registered jobs, optionally filtered by their current state.
get_job_state

Return the current run state registered for a job.
get_job_state_datetime

Return when the job was recorded in a specific run state.
set_job_state

Register a job transition, update indexes, and emit the matching log event.
Source code in src/omnipy/shared/protocols/hub/registry.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@runtime_checkable
class IsRunStateRegistry(Protocol):
    """Protocol for registries that track job run states."""
    def __init__(self) -> None:
        ...

    def get_job_state(self, job: IsUniquelyNamedJob) -> RunState.Literals:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_SUMMARY}}
        #
        # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DETAILS}}
        """Return the current run state registered for a job.

        Args:
            job: Job whose current run state should be looked up.

        Returns:
            RunState.Literals: Current run-state literal for the job.
        """
        ...

    def get_job_state_datetime(self, job: IsUniquelyNamedJob, state: RunState.Literals) -> datetime:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DATETIME_SUMMARY}}
        #
        # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DATETIME_DETAILS}}
        """Return when the job was recorded in a specific run state.

        Args:
            job: Job whose transition time should be looked up.
            state: Run-state literal to query.

        Returns:
            datetime: Timestamp recorded for the requested transition.
        """
        ...

    def all_jobs(self,
                 state: RunState.Literals | None = None) -> tuple[IsUniquelyNamedJob, ...]:  # noqa
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNSTATEREGISTRY_ALL_JOBS_SUMMARY}}
        #
        # {{ISRUNSTATEREGISTRY_ALL_JOBS_DETAILS}}
        """Return all registered jobs, optionally filtered by their current state.

        Args:
            state: Optional run-state filter limiting the returned jobs.

        Returns:
            tuple[IsUniquelyNamedJob, ...]: Registered jobs matching the requested filter.
        """
        ...

    def set_job_state(self, job: IsUniquelyNamedJob, state: RunState.Literals) -> None:
        # %% Original docstring (managed by expand_docstr_macros.py) %%
        # {{ISRUNSTATEREGISTRY_SET_JOB_STATE_SUMMARY}}
        #
        # {{ISRUNSTATEREGISTRY_SET_JOB_STATE_DETAILS}}
        """Register a job transition, update indexes, and emit the matching log event.

        Args:
            job: Job whose state transition should be recorded.
            state: New run-state literal to register.
        """
        ...

__init__ 

__init__() -> None
Source code in src/omnipy/shared/protocols/hub/registry.py 
56
57
def __init__(self) -> None:
    ...

all_jobs 

all_jobs(state: RunState.Literals | None = None) -> tuple[IsUniquelyNamedJob, ...]

Return all registered jobs, optionally filtered by their current state.

PARAMETER DESCRIPTION
state

Optional run-state filter limiting the returned jobs.

TYPE: RunState.Literals | None DEFAULT: None

RETURNS DESCRIPTION
tuple[IsUniquelyNamedJob, ...]

tuple[IsUniquelyNamedJob, ...]: Registered jobs matching the requested filter.
Source code in src/omnipy/shared/protocols/hub/registry.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def all_jobs(self,
             state: RunState.Literals | None = None) -> tuple[IsUniquelyNamedJob, ...]:  # noqa
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNSTATEREGISTRY_ALL_JOBS_SUMMARY}}
    #
    # {{ISRUNSTATEREGISTRY_ALL_JOBS_DETAILS}}
    """Return all registered jobs, optionally filtered by their current state.

    Args:
        state: Optional run-state filter limiting the returned jobs.

    Returns:
        tuple[IsUniquelyNamedJob, ...]: Registered jobs matching the requested filter.
    """
    ...

get_job_state 

get_job_state(job: IsUniquelyNamedJob) -> RunState.Literals

Return the current run state registered for a job.

PARAMETER DESCRIPTION
job

Job whose current run state should be looked up.

TYPE: IsUniquelyNamedJob

RETURNS DESCRIPTION
RunState.Literals

RunState.Literals: Current run-state literal for the job.
Source code in src/omnipy/shared/protocols/hub/registry.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_job_state(self, job: IsUniquelyNamedJob) -> RunState.Literals:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_SUMMARY}}
    #
    # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DETAILS}}
    """Return the current run state registered for a job.

    Args:
        job: Job whose current run state should be looked up.

    Returns:
        RunState.Literals: Current run-state literal for the job.
    """
    ...

get_job_state_datetime 

get_job_state_datetime(job: IsUniquelyNamedJob, state: RunState.Literals) -> datetime

Return when the job was recorded in a specific run state.

PARAMETER DESCRIPTION
job

Job whose transition time should be looked up.

TYPE: IsUniquelyNamedJob

state

Run-state literal to query.

TYPE: RunState.Literals

RETURNS DESCRIPTION
datetime

Timestamp recorded for the requested transition.

TYPE: datetime

Source code in src/omnipy/shared/protocols/hub/registry.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_job_state_datetime(self, job: IsUniquelyNamedJob, state: RunState.Literals) -> datetime:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DATETIME_SUMMARY}}
    #
    # {{ISRUNSTATEREGISTRY_GET_JOB_STATE_DATETIME_DETAILS}}
    """Return when the job was recorded in a specific run state.

    Args:
        job: Job whose transition time should be looked up.
        state: Run-state literal to query.

    Returns:
        datetime: Timestamp recorded for the requested transition.
    """
    ...

set_job_state 

set_job_state(job: IsUniquelyNamedJob, state: RunState.Literals) -> None

Register a job transition, update indexes, and emit the matching log event.

PARAMETER DESCRIPTION
job

Job whose state transition should be recorded.

TYPE: IsUniquelyNamedJob

state

New run-state literal to register.

TYPE: RunState.Literals

Source code in src/omnipy/shared/protocols/hub/registry.py 
106
107
108
109
110
111
112
113
114
115
116
117
def set_job_state(self, job: IsUniquelyNamedJob, state: RunState.Literals) -> None:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # {{ISRUNSTATEREGISTRY_SET_JOB_STATE_SUMMARY}}
    #
    # {{ISRUNSTATEREGISTRY_SET_JOB_STATE_DETAILS}}
    """Register a job transition, update indexes, and emit the matching log event.

    Args:
        job: Job whose state transition should be recorded.
        state: New run-state literal to register.
    """
    ...