Skip to content

cmem_client.repositories.workflows¤

Repository for the workflows of DataIntegration projects.

Provides WorkflowsRepository for listing workflows and for starting them, polling their execution status and running them with input or output payloads.

Examples:

List the workflows of the deployment:

>>> from cmem_client.client import Client
>>> client = Client.from_env()
>>> for workflow_id in client.workflows:
...     print(workflow_id, client.workflows[workflow_id].label)

Start a workflow and wait until it finished:

>>> client.workflows.execute("my-project:my-workflow")
>>> client.workflows.get_status("my-project:my-workflow")
>>> client.workflows.execute_wait_for_completion("my-project:my-workflow")

Look at the status of every workflow currently known:

>>> client.workflows.get_all_statuses()

Classes:

WorkflowsRepository¤

Bases: TaskSearchRepository

Repository for managing workflows in Corporate Memory.

The dict (keys, values, items) is populated via the task search API which returns workflows with io info (variableInputs/variableOutputs) and tags in a single call. Operational methods (execute, get_status, execute_io, etc.) are independent of the dict and hit dedicated activity/result endpoints.

Functions:

  • execute – Execute the workflow without waiting for completion.
  • execute_io – Execute a workflow with variable input/output as a streaming context manager.
  • execute_wait_for_completion – Execute the workflow and block until it finishes.
  • fetch_data – Fetch a list from the DI task search endpoint via a type adapter.
  • get_all_statuses – Get status information for multiple workflow activities.
  • get_status – Get the current status of a workflow activity.
  • get_task – Get full task details from the API.
  • get_workflow_editor_url – Get the URL to open a workflow in the workbench editor.
  • items – Get the items of the repository
  • keys – Get the keys of the repository
  • values – Get the values of the repository

Attributes:

  • logger (Logger) – Gets the client logger

execute¤

execute(workflow_id, activity_name='ExecuteDefaultWorkflow')

Execute the workflow without waiting for completion.

Parameters:

  • workflow_id (str) – The workflow to execute (in the form of ‘project_id:workflow_id’)
  • activity_name (ACTIVITY_NAME) – Name of the activity

Raises:

execute_io¤

execute_io(workflow_id, input_file=None, input_mime_type='application/xml', output_mime_type='application/xml', auto_config=False)

Execute a workflow with variable input/output as a streaming context manager.

Parameters:

  • workflow_id (str) – Workflow ID in the form ‘project_id:task_id’.
  • input_file (str | None) – Optional path to the input file.
  • input_mime_type (str) – MIME type of the input file.
  • output_mime_type (str) – MIME type expected for the output.
  • auto_config (bool) – Whether to enable auto-configuration of input datasets.

Yields:

  • Generator[Response] – httpx.Response: Streaming response from the workflow execution.

Raises:

execute_wait_for_completion¤

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

Execute the workflow and block until it finishes.

Parameters:

  • workflow_id (str) – The workflow to execute (in the form of ‘project_id:workflow_id’)
  • activity_name (ACTIVITY_NAME) – Activity name. Defaults to “ExecuteDefaultWorkflow”.
  • sleep_time (int) – Seconds to sleep between status polls. Defaults to 1.

Raises:

fetch_data¤

fetch_data()

Fetch a list from the DI task search endpoint via a type adapter.

get_all_statuses¤

get_all_statuses(project_id=None, status_filter=None, activity_type=None)

Get status information for multiple workflow activities.

Parameters:

  • project_id (str | None) – Optional project ID to filter by.
  • status_filter (str | None) – Optional status filter (e.g. “Finished”, “Running”).
  • activity_type (str | None) – Optional activity type to filter by (e.g. “ExecuteDefaultWorkflow”).

Returns:

Raises:

get_status¤

get_status(workflow_id, activity_name='ExecuteDefaultWorkflow')

Get the current status of a workflow activity.

Parameters:

  • workflow_id (str) – Workflow ID in the form ‘project_id:workflow_id’.
  • activity_name (ACTIVITY_NAME) – Activity to check. Defaults to “ExecuteDefaultWorkflow”.

Returns:

  • WorkflowStatus – WorkflowStatus with current state, progress, and message.

Raises:

get_task¤

get_task(project_id, task_id, with_labels=True)

Get full task details from the API.

Parameters:

  • project_id (str) – The project ID.
  • task_id (str) – The task ID.
  • with_labels (bool) – Whether to include labels in the response.

Returns:

  • TaskResponse – The full task details as a TaskResponse model.

get_workflow_editor_url¤

get_workflow_editor_url(workflow_id)

Get the URL to open a workflow in the workbench editor.

Parameters:

  • workflow_id (str) – Workflow ID in the form ‘project_id:task_id’.

Returns:

  • str – URL string for the workflow editor.

items¤

items()

Get the items of the repository

keys¤

keys()

Get the keys of the repository

logger¤

logger: logging.Logger

Gets the client logger

values¤

values()

Get the values of the repository

Comments