Skip to content

cmem_client.models.workflow¤

Workflow models

A workflow is the DataIntegration task which moves data between datasets. The workflows of all projects are the items of client.workflows, keyed by {project_id}:{workflow_id}.

A workflow which came out of the repository carries a client, so it can start itself and report its own status instead of going back through the repository:

>>> from cmem_client.client import Client
>>> client = Client.from_env()
>>> workflow = client.workflows["my-project:my-workflow"]
>>> workflow.execute_wait_for_completion()
>>> workflow.get_status().concrete_status

Classes:

Attributes:

ACTIVITY_NAME¤

ACTIVITY_NAME = Literal['ExecuteDefaultWorkflow', 'ExecuteLocalWorkflow', 'ExecuteWorkflowWithPayload']

ACTIVITY_TYPE_EXECUTE_DEFAULTWORKFLOW¤

ACTIVITY_TYPE_EXECUTE_DEFAULTWORKFLOW = 'ExecuteDefaultWorkflow'

ACTIVITY_TYPE_EXECUTE_LOCALWORKFLOW¤

ACTIVITY_TYPE_EXECUTE_LOCALWORKFLOW = 'ExecuteLocalWorkflow'

ACTIVITY_TYPE_EXECUTE_WITH_PAYLOAD¤

ACTIVITY_TYPE_EXECUTE_WITH_PAYLOAD = 'ExecuteWorkflowWithPayload'

VALID_WORKFLOW_STATUSES¤

VALID_WORKFLOW_STATUSES = ['Idle', 'Not executed', 'Finished', 'Cancelled', 'Failed', 'Successful', 'Canceling', 'Running', 'Waiting']

Workflow¤

Bases: Model, ReadRepositoryItem

A workflow

Attributes:

  • id (str) – ID of the workflow, unique within its project.
  • label (str) – Human readable name of the workflow.
  • project_id (str) – ID of the project holding the workflow. Together with id it forms the {project_id}:{id} key of the repository.
  • project_label (str) – Human readable name of that project.
  • variable_inputs (list[str]) – IDs of the inputs which can be replaced at execution time, which is what execute_io() writes its payload into.
  • variable_outputs (list[str]) – IDs of the outputs which can be replaced at execution time, which is what execute_io() reads its result from.
  • warnings (list[str]) – Warnings DataIntegration reported for the workflow.
  • tags (list[Tag]) – Tags attached to the workflow.
  • parameters (dict) – Raw parameters as returned by the search endpoint. Excluded from serialization; the variable inputs and outputs are read out of it.

Functions:

execute¤

execute(activity_name='ExecuteDefaultWorkflow')

Execute the workflow

execute_wait_for_completion¤

execute_wait_for_completion(activity_name='ExecuteDefaultWorkflow', sleep_time=1)

Execute the workflow waiting for completion

get_id¤

get_id()

Get the workflow ID

get_status¤

get_status(activity_name='ExecuteDefaultWorkflow')

Get the status of the workflow execution.

id¤

id: str

label¤

label: str

model_config¤

model_config = ConfigDict(extra='allow', populate_by_name=True)

parameters¤

parameters: dict = Field(default_factory=dict, exclude=True)

project_id¤

project_id: str = Field(alias='projectId')

project_label¤

project_label: str = Field(alias='projectLabel', default='')

set_client¤

set_client(client)

Set the client for this workflow

tags¤

tags: list[Tag] = Field(default_factory=list)

variable_inputs¤

variable_inputs: list[str] = Field(alias='variableInputs', default_factory=list)

variable_outputs¤

variable_outputs: list[str] = Field(alias='variableOutputs', default_factory=list)

warnings¤

warnings: list[str] = Field(default_factory=list)

WorkflowSearchResultSet¤

Bases: Model

Wrapper for the search API response envelope.

Attributes:

model_config¤

model_config = ConfigDict(extra='allow', populate_by_name=True)

results¤

results: list[Workflow]

WorkflowStatus¤

Bases: Model

Workflow execution status

Attributes:

  • status_name (str) – Coarse state of the execution, e.g. Idle, Running or Finished.
  • concrete_status (str) – What the state amounts to, e.g. Successful, Failed or Cancelled. This is the field to check once an execution finished.
  • progress (float | None) – How far the execution got, in percent, or None if the workflow does not report progress.
  • failed (bool) – Whether the execution failed.
  • message (str) – Human readable status message.
  • last_update_time (int) – When the status was last updated, as a Unix timestamp in milliseconds.
  • project (str) – ID of the project holding the workflow.
  • task (str) – ID of the workflow task.
  • activity (str) – Name of the activity which runs the workflow.
  • activity_label (str) – Human readable name of that activity.
  • queue_time (datetime | None) – When the execution was queued.
  • start_time (datetime | None) – When the execution actually started.
  • is_running (bool) – Whether the execution is still going. Poll this to wait for a workflow, or let execute_wait_for_completion() do it.
  • runtime (int | None) – How long the execution took, in milliseconds.
  • cancelled (bool | None) – Whether the execution was cancelled.
  • exception_message (str | None) – Message of the exception which ended the execution, if one did.

activity¤

activity: str

activity_label¤

activity_label: str = Field(alias='activityLabel')

cancelled¤

cancelled: bool | None = None

concrete_status¤

concrete_status: str = Field(alias='concreteStatus')

exception_message¤

exception_message: str | None = Field(default=None, alias='exceptionMessage')

failed¤

failed: bool

is_running¤

is_running: bool = Field(alias='isRunning')

last_update_time¤

last_update_time: int = Field(alias='lastUpdateTime')

message¤

message: str

model_config¤

model_config = ConfigDict(extra='allow', populate_by_name=True)

progress¤

progress: float | None

project¤

project: str

queue_time¤

queue_time: datetime | None = Field(default=None, alias='queueTime')

runtime¤

runtime: int | None = None

start_time¤

start_time: datetime | None = Field(default=None, alias='startTime')

status_name¤

status_name: str = Field(alias='statusName')

task¤

task: str

Comments