Workflow State¶

The workflow state domain provides access to observation workflow state queries and updates.

Use workflow_state to inspect workflow state and transition observations through valid workflow stages.

Quick Example¶

from gpp_client.generated import ObservationWorkflowState

async with GPPClient() as client:
   result = await client.workflow_state.update_by_id(
      observation_id="o-123",
      workflow_state=ObservationWorkflowState.READY,
   )

Retrieving Workflow State¶

Get workflow state by observation ID:

result = await client.workflow_state.get_by_id("o-123")

Get workflow state by observation reference:

result = await client.workflow_state.get_by_reference("GN-2026A-Q-1-1")

Updating Workflow State¶

Update workflow state by observation ID:

result = await client.workflow_state.update_by_id(
   observation_id="o-123",
   workflow_state=workflow_state,
)

The domain validates that:

The observation calculation state is ready
The requested transition is valid
The workflow state is not already set

Retrying Workflow Updates¶

Use update_by_id_with_retry to retry while background calculation is still in progress:

result = await client.workflow_state.update_by_id_with_retry(
   observation_id="o-123",
   workflow_state=workflow_state,
   max_attempts=10,
   retry_delay=1.0,
)

This is useful when workflow updates depend on calculation state becoming ready.

Error Handling¶

Workflow state updates may raise:

GPPRetryableError when calculation state is not ready
GPPValidationError for invalid workflow transitions
GPPClientError for other client-side failures

Notes¶

All workflow state operations use GraphQL.

The retry helper only retries the not-ready case. Validation and other client errors fail immediately.

API Reference¶

class gpp_client.domains.workflow_state.WorkflowStateDomain(*, graphql: GraphQLClient, rest: RESTClient, settings: GPPSettings)[source]¶

Bases: BaseDomain

Domain for managing observation workflow states.

async get_by_id(observation_id: str) → GetObservationWorkflowStateById[source]¶

Get workflow state details for an observation by ID.

Parameters:: observation_id (str) – The observation ID.
Returns:: The generated GraphQL response model.
Return type:: GetObservationWorkflowStateById

async get_by_reference(observation_reference: str) → GetObservationWorkflowStateByReference[source]¶

Get workflow state details for an observation by reference.

Parameters:: observation_reference (str) – The observation reference label.
Returns:: The generated GraphQL response model.
Return type:: GetObservationWorkflowStateByReference

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

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:

observation_id (str) – The observation ID.
workflow_state (ObservationWorkflowState) – The desired workflow state to transition to.

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(observation_id: str, *, workflow_state: ObservationWorkflowState, max_attempts: int = 10, initial_delay: float = 0.0, retry_delay: float = 1.0) → dict[str, Any][source]¶

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.