Workflow State

Manager for interacting with observation workflow state resources.

class gpp_client.managers.workflow_state.WorkflowStateManager(client: GPPClient)

Bases: BaseManager

Manager for interacting with observation workflow state resources.

async get_by_id(*, observation_id: str | None = None, observation_reference: str | None = None) → dict[str, Any]

Get the workflow state and calculation state of an observation by its ID or reference.

Parameters:

• observation_id (Optional[str], optional) – The observation ID, by default None.

• observation_reference (Optional[str], optional) – The observation reference, by default None.

Returns:

The returned workflow state for the observation.

Return type:

dict[str, Any]

Raises:

• GPPValidationError – If neither or both of observation_id and observation_reference are provided.

• GPPClientError – If neither or both of observation_id and observation_reference are provided, or if the observation cannot be found.

get_result(result: dict[str, Any] | None, operation_name: str | None = None) → dict[str, Any]

Extract the payload for a given GraphQL operation.

Parameters:

• result (dict[str, Any] | None) – The full GraphQL result.

• operation_name (str | None, optional) – The name of the operation to extract. If None, and the result contains exactly one top-level key, that key is used.

Returns:

The extracted payload for the specified operation.

Return type:

dict[str, Any]

Raises:

GPPClientError – If the result is empty or invalid, or if the specified operation name is not found in the result.

get_single_result(payload: dict[str, Any], key: str) → dict[str, Any]

Extract exactly one item from a list-valued field in a GraphQL payload.

Parameters:

• payload (dict[str, Any]) – The GraphQL payload containing the field.

• key (str) – The key of the field to extract.

Returns:

The single item extracted from the field.

Return type:

dict[str, Any]

Raises:

GPPClientError – If the specified field is missing from the payload, is not a list, or does not contain exactly one item.

load_properties(*, properties: Any | None, from_json: str | Path | dict[str, Any] | None, cls: type[T]) → T

Return a validated properties object from exactly one data source.

Parameters:

• properties (T, optional) – Preconstructed properties instance. Returned unchanged when provided.

• from_json (str | Path | dict[str, Any], optional) – Path to a JSON file or a dictionary containing the JSON data.

• cls (type[T]) – Concrete PropertiesInput class for validation. Required.

Returns:

Instance of cls representing the validated properties.

Return type:

T

Raises:

GPPValidationError – If validation fails or an error occurs loading the properties.

raise_error(exc_class: type[GPPError], exc: Exception, *, include_traceback: bool = False) → NoReturn

Raise a structured exception with contextual information.

Parameters:

• exc_class (type[GPPError]) – The exception class to raise (must accept a string message).

• exc (Exception) – The original exception to wrap.

• include_traceback (bool, default=False) – Whether to include the original traceback using from exc.

Raises:

type[GPPError] – The raised exception of the specified type.

async raise_for_status(response, *, ok_statuses: set[int], default_message: str = 'Request failed') → None

Raise an exception if the HTTP response status is not OK.

Parameters:

• response (aiohttp.ClientResponse) – The HTTP response to check.

• ok_statuses (set[int]) – Set of acceptable HTTP status codes.

• default_message (str, default="Request failed") – Default error message if the response has no content.

Raises:

GPPResponseError – If the response status is not in ok_statuses.

resolve_content(*, file_path: str | Path | None, content: bytes | None) → bytes

Resolve upload content bytes from exactly one source.

Parameters:

• file_path (str | Path | None) – Path to a local file. If provided, it must exist and be a file.

• content (bytes | None) – Raw bytes content.

Returns:

The resolved content bytes.

Return type:

bytes

Raises:

• GPPValidationError – If both or neither of file_path and content are provided, or if file_path is invalid.

• GPPClientError – If reading the file fails due to an unexpected I/O error.

async update_by_id(*, workflow_state: ObservationWorkflowState, observation_id: str) → dict[str, Any]

Update the workflow state of an observation by its ID, or return the current workflow if already set.

This function will:

• Fetch the current observation and its workflow.

• If the calculation state is not READY, raise an error to retry later.

• If the desired state is already set, return the workflow as-is.

• Otherwise, validate the requested workflow state against validTransitions.

• If valid, submit the mutation to update the workflow state.

Parameters:

• workflow_state (ObservationWorkflowState) – The desired workflow state to transition to.

• observation_id (str) – The observation ID.

Returns:

The returned workflow state for the observation.

Return type:

dict[str, Any]

Raises:

• GPPClientError – If there are general client-side errors.

• GPPValidationError – If the requested workflow state transition is invalid.

• GPPRetryableError – If the observation calculation is not READY.

async update_by_id_with_retry(*, workflow_state: ObservationWorkflowState, observation_id: str, max_attempts: int = 10, initial_delay: float = 0.0, retry_delay: float = 1.0) → dict[str, Any]

Update the workflow state of an observation by its ID, retrying if the observation is not ready.

This function wraps update_by_id with retry logic to handle cases where the observation calculation is not yet in the READY state.

Parameters:

• workflow_state (ObservationWorkflowState) – The desired workflow state to transition to.

• observation_id (str) – The observation ID.

• max_attempts (int, default=10) – Maximum number of retry attempts.

• initial_delay (float, default=0.0) – Initial delay in seconds before first attempt.

• retry_delay (float, default=1.0) – Delay in seconds between retry attempts.

Returns:

The returned workflow state for the observation.

Return type:

dict[str, Any]

Raises:

• GPPClientError – If the maximum number of retry attempts is exceeded without success.

• GPPValidationError – If the requested workflow state transition is invalid.

validate_single_identifier(**kwargs) → None

Validate that exactly one identifier is provided.

This helper checks that exactly one of the provided keyword arguments is non-None. It raises a ValueError otherwise.

Parameters:

**kwargs (dict[str, Optional[str]]) – A dictionary of identifier keyword arguments to validate.

Raises:

GPPValidationError – If none or more than one identifiers are provided.