Workflows resource

Note: Workflows with only standard steps are supported.

Import the workflow resource module and instantiate a workflow-service client

from collections.abc import Iterator
from typing import Any

from peak.resources import workflows

wf_client: workflows.Workflow = workflows.get_client()

List all existing workflow(s) and iterate over them

list_workflows_iterator: Iterator[dict[str, Any]] = wf_client.list_workflows(
    name="test",
    workflow_status=["Available"],
    last_execution_status=["Running"],
    last_modified_by=["test@peak.ai"],
)
list_workflows_iterated: dict[str, Any] = next(list_workflows_iterator)

Prepare the payload and submit a new workflow resource for creation

cron_workflow_body: dict[str, Any] = {
    "name": "cron-workflow",
    "triggers": [
        {
            "cron": "0 0 * * *",
        },
    ],
    "watchers": [
        {
            "user": "abc@peak.ai",
            "events": {
                "success": False,
                "fail": True,
            },
        },
        {
            "webhook": {
                "name": "info",
                "url": "https://abc.com/post",
                "payload": '{ "pingback-url": "https:/workflow/123" }',
            },
            "events": {
                "success": False,
                "fail": True,
                "runtimeExceeded": 10,
            },
        },
    ],
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 123,
            "imageVersionId": 1,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 21,
                "storage": "10GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "https://github.com/org/repo",
            },
        },
    },
}

cron_workflow_id: int = wf_client.create_workflow(cron_workflow_body)["id"]

webhook_workflow_body: dict[str, Any] = {
    "name": "webhook-workflow",
    "triggers": [
        {
            "webhook": True,
            "webhookPolicy": "generate",
        },
    ],
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 123,
            "imageVersionId": 1,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 23,
                "storage": "20GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "github.com/org/repo",
            },
        },
    },
}

webhook_workflow_id: int = wf_client.create_workflow(webhook_workflow_body)["id"]

manual_workflow_body: dict[str, Any] = {
    "name": "manual-workflow",
    "triggers": [],
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 123,
            "imageVersionId": 1,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 23,
                "storage": "20GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "https://github.com/org/repo",
            },
        },
    },
}

manual_workflow_id: int = wf_client.create_workflow(manual_workflow_body)["id"]

Modify the payload and update the existing workflow resource

updated_body: dict[str, Any] = {
    "name": "updated-workflow",
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 456,
            "imageVersionId": 2,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 21,
                "storage": "10GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
        },
    },
}

wf_client.update_workflow(workflow_id=manual_workflow_id, body=updated_body)

Modify the payload and update the existing workflow resource via patch-workflow

## 1. By passing the keys to update through arguments.

patch_workflow_via_args = wf_client.patch_workflow(
    workflow_id=manual_workflow_id,
    name="updated-workflow",
    image_id=100,
    image_version_id=100,
    instance_type_id=23,
    step_names=["step1"],
)


## 2. By passing body containing keys to update

body: dict[str, Any] = {
    "name": "updated-workflow",
    "triggers": [
        {
            "cron": "0 0 * * *",
        },
    ],
    "watchers": [
        {
            "user": "abc@peak.ai",
            "events": {
                "success": False,
                "fail": True,
            },
        },
        {
            "webhook": {
                "name": "info",
                "url": "https://abc.com/post",
                "payload": '{ "pingback-url": "https:/workflow/123" }',
            },
            "events": {
                "success": False,
                "fail": True,
                "runtimeExceeded": 10,
            },
        },
    ],
    "steps": {
        "step1": {
            "imageId": 100,
            "imageVersionId": 100,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 21,
                "storage": "10GB",
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "https://github.com/org/repo",
            },
        },
        "step2": {
            "imageId": 100,
            "imageVersionId": 100,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 22,
                "storage": "20GB",
            },
        },
    },
}

wf_client.patch_workflow(workflow_id=cron_workflow_id, body=body)

Create or update a workflow resource

cron_workflow_body: dict[str, Any] = {
    "name": "cron-workflow",
    "triggers": [
        {
            "cron": "0 0 * * *",
        },
    ],
    "watchers": [
        {
            "user": "abc@peak.ai",
            "events": {
                "success": False,
                "fail": True,
            },
        },
        {
            "webhook": {
                "name": "info",
                "url": "https://abc.com/post",
                "payload": '{ "pingback-url": "https:/workflow/123" }',
            },
            "events": {
                "success": False,
                "fail": True,
                "runtimeExceeded": 10,
            },
        },
    ],
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 123,
            "imageVersionId": 1,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 21,
                "storage": "10GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "https://github.com/org/repo",
            },
        },
    },
}

cron_workflow_id: int = wf_client.create_or_update_workflow(cron_workflow_body)["id"]

updated_body: dict[str, Any] = {
    "name": "cron-workflow",
    "triggers": [
        {
            "webhook": True,
            "webhookPolicy": "generate",
        },
    ],
    "tags": [
        {
            "name": "foo",
        },
        {
            "name": "bar",
        },
    ],
    "steps": {
        "stepName": {
            "type": "standard",
            "imageId": 123,
            "imageVersionId": 1,
            "command": "python script.py",
            "resources": {
                "instanceTypeId": 23,
                "storage": "20GB",
            },
            "parents": [],
            "stepTimeout": 30,
            "clearImageCache": True,
            "parameters": {
                "env": {
                    "key": "value",
                },
                "secrets": [
                    "secret-1",
                    "secret-2",
                ],
            },
            "repository": {
                "branch": "main",
                "token": "github",
                "url": "github.com/org/repo",
            },
        },
    },
}

webhook_workflow_id: int = wf_client.create_or_update_workflow(updated_body)

Describe an existing workflow resource

workflow_description: dict[str, Any] = wf_client.describe_workflow(workflow_id=manual_workflow_id)

Delete an existing workflow resource

wf_client.delete_workflow(workflow_id=63153)

List all executions of a workflow resource and iterate over them

Execution history is not available for workflow executions older than 90 days.

workflow_executions_iterator: Iterator[dict[str, Any]] = wf_client.list_executions(workflow_id=manual_workflow_id)
workflow_executions_iterated: dict[str, Any] = next(workflow_executions_iterator)

Get Details about a workflow execution

workflow_execution = wf_client.get_execution_details(
    workflow_id=1,
    execution_id="execution_id",
)

Get logs for a specific execution of a workflow resource

workflow_execution_logs = wf_client.get_execution_logs(
    workflow_id=1,
    execution_id="execution_id",
    step_name="step_name",
)

# To get next set of logs, pass the nextToken returned in the response of previous function call
next_set_of_workflow_execution_logs = wf_client.get_execution_logs(
    workflow_id=1,
    execution_id="execution_id",
    step_name="step_name",
    next_token=workflow_execution_logs["nextToken"],
)

# To download the logs locally the `save` parameter can be used along with `file_name`
next_set_of_workflow_execution_logs = wf_client.get_execution_logs(
    workflow_id=1,
    execution_id="execution_id",
    step_name="step_name",
    save=True,
    file_name="workflow_execution_logs.log",
)

Schedule an immediate execution for a workflow

wf_client.execute_workflow(workflow_id=manual_workflow_id)

# Execute a workflow with some parameters
body = {
    "params": {
        "global": {
            "key": "value",
        },
        "stepWise": {
            "step1": {
                "key1": "value1",
            },
        },
    },
}

wf_client.execute_workflow(workflow_id=manual_workflow_id, body=body)

Get all available resources for workflow

available_resources: list[dict[str, Any]] = wf_client.list_resources()

Get default resource values for workflow

default_resource: dict[str, Any] = wf_client.get_default_resource()