radical.asyncflow.constants¶

@classmethod
def register_backend_states(
    cls,
    backend: Any,
    done_state: Any,
    failed_state: Any,
    canceled_state: Any,
    running_state: Any,
    **additional_states) -> None:
    """Register state mappings for a new backend.

    Associates a backend identifier with its corresponding task state values,
    mapping the standard task states to backend-specific representations.
    Supports additional custom state mappings beyond the four core states.

    Args:
        backend (Any): The identifier for the backend to register. Can be
            a string, module, or any hashable object.
        done_state (Any): Backend's representation of the DONE state.
        failed_state (Any): Backend's representation of the FAILED state.
        canceled_state (Any): Backend's representation of the CANCELED state.
        running_state (Any): Backend's representation of the RUNNING state.
        **additional_states: Additional state mappings where the key is the
            state name (str) and the value is the backend's representation.
            Keys will be converted to uppercase and matched against TasksMainStates.

    Returns:
        None

    Example:
        ::
            StateMapper.register_backend_states(
                backend='slurm',
                done_state='COMPLETED',
                failed_state='FAILED',
                canceled_state='CANCELLED',
                running_state='RUNNING',
                pending='PENDING',
                timeout='TIMEOUT'
            )
    """
    cls._backend_registry[backend] = {
        TasksMainStates.DONE: done_state,
        TasksMainStates.FAILED: failed_state,
        TasksMainStates.CANCELED: canceled_state,
        TasksMainStates.RUNNING: running_state,
        **{TasksMainStates[k.upper()]: v for k, v in additional_states.items()}}

@classmethod
def register_backend_states_with_defaults(cls, backend: Any):
    """Register a backend using default main state values.

    Convenience method that registers a backend where the backend-specific
    states are identical to the main state values (i.e., the string values
    of the TasksMainStates enum).

    Args:
        backend (Any): The backend identifier to register.

    Returns:
        The result of register_backend_states() with default values.

    Example:
        ::
            # This registers backend states as:
            # DONE -> "DONE", FAILED -> "FAILED", etc.
            StateMapper.register_backend_states_with_defaults('thread_backend')
    """
    return cls.register_backend_states(backend,
                                       done_state=TasksMainStates.DONE.value,
                                       failed_state=TasksMainStates.FAILED.value,
                                       canceled_state=TasksMainStates.CANCELED.value,
                                       running_state=TasksMainStates.RUNNING.value)

def __getattr__(self, name: str) -> Any:
    """Access backend-specific states directly via attribute notation.

    Enables direct access to backend states using the main state names
    as attributes (e.g., mapper.DONE, mapper.FAILED).

    Args:
        name (str): The main state name to access (DONE, FAILED, CANCELED, RUNNING).

    Returns:
        Any: The backend-specific state value corresponding to the main state.

    Raises:
        AttributeError: If the specified state name is not valid.

    Example:
        ::
            mapper = StateMapper('my_backend')
            done_state = mapper.DONE  # Returns backend's DONE state
            running_state = mapper.RUNNING  # Returns backend's RUNNING state
    """
    try:
        main_state = TasksMainStates[name]
        return self._state_map[main_state]
    except KeyError:
        raise AttributeError(f"'{self.__class__.__name__}' has no state '{name}'")

def to_main_state(self, backend_state: Any) -> TasksMainStates:
    """Convert backend-specific state to main state.

    Translates a backend-specific state value back to the corresponding
    TasksMainStates enum value.

    Args:
        backend_state (Any): The backend-specific state value to convert.

    Returns:
        TasksMainStates: The corresponding main state enum value.

    Raises:
        ValueError: If the backend state is not recognized.

    Example:
        ::
            mapper = StateMapper('slurm')
            main_state = mapper.to_main_state('COMPLETED')  # Returns TasksMainStates.DONE
    """
    try:
        return self._reverse_map[backend_state]
    except KeyError:
        raise ValueError(f"Unknown backend state: {backend_state}")

def get_backend_state(self, main_state: Union[TasksMainStates, str]) -> Any:
    """Get backend-specific state for a main state.

    Retrieves the backend-specific state value that corresponds to the
    given main state. Accepts both TasksMainStates enum values and
    string representations.

    Args:
        main_state (Union[TasksMainStates, str]): The main state to convert.
            Can be a TasksMainStates enum value or its string representation.

    Returns:
        Any: The backend-specific state value.

    Raises:
        KeyError: If the main state is not found in the mapping.

    Example:
        ::
            mapper = StateMapper('my_backend')
            backend_state = mapper.get_backend_state(TasksMainStates.DONE)
            # Or using string
            backend_state = mapper.get_backend_state('DONE')
    """
    if isinstance(main_state, str):
        main_state = TasksMainStates[main_state]
    return self._state_map[main_state]