Upsert Workflows

async AsyncCogniteClient.workflows.versions.upsert(

version: WorkflowVersionUpsert | Sequence[WorkflowVersionUpsert],

mode: Literal['replace'] = 'replace',

) → WorkflowVersion | WorkflowVersionList

Create one or more workflow version(s).

Note this is an upsert endpoint, so workflow versions that already exist will be updated, and new ones will be created.

Parameters:

• version (WorkflowVersionUpsert | Sequence[WorkflowVersionUpsert]) – The workflow version(s) to upsert.

• mode (Literal['replace']) – This is not an option for the API, but is included here to document that the upserts are always done in replace mode.

Returns:

The created workflow version(s).

Return type:

WorkflowVersion | WorkflowVersionList

Examples

Create one workflow version with a single Function task:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import (
...     WorkflowVersionUpsert,
...     WorkflowDefinitionUpsert,
...     WorkflowTask,
...     FunctionTaskParameters,
... )
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> function_task = WorkflowTask(
...     external_id="my_workflow-task1",
...     parameters=FunctionTaskParameters(
...         external_id="my_fn_xid",
...         data={"a": 1, "b": 2},
...     ),
... )
>>> new_version = WorkflowVersionUpsert(
...     workflow_external_id="my_workflow",
...     version="1",
...     workflow_definition=WorkflowDefinitionUpsert(
...         tasks=[function_task],
...         description="This workflow has one step",
...     ),
... )
>>> res = client.workflows.versions.upsert(new_version)