Press Blocks

Import the press block module and Instantiate a press-block client

from collections.abc import Iterator
from typing import Any

from peak.press import blocks

client: blocks.Block = blocks.get_client()

Prepare payload and create a new workflow block-spec

workflow_private_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "workflow",
    "metadata": {
        "name": "workflow-block-private",
        "title": "Workflow Block",
        "summary": "Creating a new workflow block spec.",
        "description": "Creating a new workflow block spec from SDK.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "tags": [
            {
                "name": "sdk",
            },
        ],
    },
    "release": {
        "version": "1.0.0",
        "notes": "This is the original release",
    },
    "config": {
        "triggers": [  ## Time Based trigger
            {
                "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,
                },
            },
        ],
        "images": {
            "example-image": {
                "buildArguments": {
                    "CHUNK_SIZE": "1000",
                },
                "context": ".",
                "dockerfile": "Dockerfile",
                "secrets": ["SECRET_1"],
                "useCache": False,
                "version": "0.0.1",
            },
        },
        "steps": {
            "step-1": {
                "type": "standard",
                "imageRef": "example-image",
                "resources": {
                    "instanceTypeId": 21,
                    "storage": "10GB",
                },
                "command": "python main.py",
            },
            "step-2": {
                "type": "standard",
                "imageDetails": {
                    "id": 1,
                    "versionId": 1,
                },
                "resources": {
                    "instanceTypeId": 21,
                    "storage": "10GB",
                },
                "command": "python main.py",
            },
        },
    },
}

workflow_parameters: dict[str, Any] = {
    "run": [
        {
            "defaultValue": "AVG",
            "description": "Select an aggregation function (e.g., AVG, SUM, COUNT)",
            "name": "agg_type",
            "required": False,
            "title": "Agg Type",
            "type": "string",
        },
        {
            "defaultValue": False,
            "description": "Enable email notifications",
            "name": "email_notifications",
            "required": False,
            "title": "Email Notifications",
            "type": "boolean",
        },
        {
            "defaultValue": 10,
            "description": "Select the number of iterations",
            "name": "num_iterations",
            "options": [
                {
                    "title": "Low",
                    "value": 10,
                },
                {
                    "title": "Medium",
                    "value": 50,
                },
                {
                    "title": "High",
                    "value": 100,
                },
            ],
            "required": False,
            "title": "Number of Iterations",
            "type": "number",
        },
        {
            "defaultValue": ["input.csv", "output.csv"],
            "description": "Specify input and output file names",
            "name": "file_names",
            "required": True,
            "title": "File Names",
            "type": "string_array",
        },
    ],
}

workflow_private_spec: dict[str, str] = client.create_spec(
    body=workflow_private_spec_body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    featured=True,
    scope="private",
    parameters=workflow_parameters,
)

workflow_shared_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "workflow",
    "metadata": {
        "name": "workflow-block-shared",
        "title": "Workflow Block",
        "summary": "Creating a new workflow block spec.",
        "description": "Creating a new workflow block spec from SDK.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "tags": [
            {
                "name": "sdk",
            },
        ],
    },
    "release": {
        "version": "1.0.0",
        "notes": "This is the original release",
    },
    "config": {
        "triggers": [  ## Webhook Based trigger
            {
                "webhook": True,
            },
        ],
        "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": {
            "step-1": {
                "type": "standard",
                "image": {
                    "context": ".",
                    "dockerfile": "Dockerfile",
                    "version": "0.0.1",
                },
                "resources": {
                    "instanceTypeId": 21,
                    "storage": "10GB",
                },
                "command": "python main.py",
            },
        },
    },
}

workflow_shared_spec: dict[str, str] = client.create_spec(
    body=workflow_shared_spec_body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    featured=True,
    scope="shared",
    tenants=["tenant1", "tenant2"],
)

Prepare payload and create a new webapp block-spec

webapp_private_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "webapp",
    "metadata": {
        "name": "webapp-block-private",
        "title": "Webapp Block",
        "summary": "Creating a new webapp block spec.",
        "description": "Creating a new webapp block spec from SDK.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "tags": [
            {
                "name": "sdk",
            },
        ],
    },
    "release": {
        "version": "1.0.0",
        "notes": "This is the original release",
    },
    "config": {
        "imageDetails": {
            "id": 123,
            "versionId": 1,
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "sessionStickiness": True,
    },
}

webapp_parameters: dict[str, Any] = {
    "run": [
        {
            "defaultValue": "AVG",
            "description": "Select an aggregation function (e.g., AVG, SUM, COUNT)",
            "name": "agg_type",
            "required": False,
            "title": "Agg Type",
            "type": "string",
        },
        {
            "defaultValue": False,
            "description": "Enable email notifications",
            "name": "email_notifications",
            "required": False,
            "title": "Email Notifications",
            "type": "boolean",
        },
        {
            "defaultValue": 10,
            "description": "Select the number of iterations",
            "name": "num_iterations",
            "options": [
                {
                    "title": "Low",
                    "value": 10,
                },
                {
                    "title": "Medium",
                    "value": 50,
                },
                {
                    "title": "High",
                    "value": 100,
                },
            ],
            "required": False,
            "title": "Number of Iterations",
            "type": "number",
        },
        {
            "defaultValue": ["input.csv", "output.csv"],
            "description": "Specify input and output file names",
            "name": "file_names",
            "required": True,
            "title": "File Names",
            "type": "string_array",
        },
    ],
}

webapp_private_spec: dict[str, str] = client.create_spec(
    body=webapp_private_spec_body,
    featured=True,
    scope="private",
    parameters=webapp_parameters,
)

webapp_shared_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "webapp",
    "metadata": {
        "name": "webapp-block-shared",
        "title": "Webapp Block",
        "summary": "Creating a new webapp block spec.",
        "description": "Creating a new webapp block spec from SDK.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "tags": [
            {
                "name": "sdk",
            },
        ],
    },
    "release": {
        "version": "1.0.0",
        "notes": "This is the original release",
    },
    "config": {
        "image": {
            "buildArguments": {
                "CHUNK_SIZE": "1000",
            },
            "context": ".",
            "dockerfile": "Dockerfile",
            "version": "0.0.1",
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "sessionStickiness": True,
    },
}

webapp_shared_spec: dict[str, str] = client.create_spec(
    body=webapp_shared_spec_body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    featured=True,
    scope="shared",
    tenants=["tenant1", "tenant2"],
)

List created block-spec(s) and iterate over them

spec_iterator: Iterator[dict[str, Any]] = client.list_specs()
spec_iterated: dict[str, Any] = next(spec_iterator)

Describe an already existing block-spec

existing_spec: dict[str, Any] = client.describe_spec(workflow_private_spec["id"])

Update the metadata and discoverability of a block-spec

updated_spec_body: dict[str, Any] = {
    "metadata": {
        "name": "updated-block-spec",
        "title": "Updated Spec Metadata",
        "summary": "Updating block spec metadata.",
        "description": "Updating block spec metadata from SDK.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "tags": [
            {
                "name": "sdk",
            },
        ],
        "status": "available",
    },
    "featured": False,
    "scope": "private",
}

client.update_spec_metadata(
    spec_id=workflow_private_spec["id"],
    body=updated_spec_body,
)

Create a new workflow block-spec release with updated config and release

body: dict[str, Any] = {
    "config": {
        "triggers": [],  ## Manual trigger
        "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,
                },
            },
        ],
        "images": {
            "example-image": {
                "buildArguments": {
                    "CHUNK_SIZE": "1000",
                },
                "context": ".",
                "dockerfile": "Dockerfile",
                "secrets": ["SECRET_1"],
                "useCache": False,
                "version": "0.0.1",
            },
        },
        "steps": {
            "step-1": {
                "type": "standard",
                "imageRef": "example-image",
                "resources": {
                    "instanceTypeId": 21,
                    "storage": "10GB",
                },
                "command": "python main.py",
            },
            "step-2": {
                "type": "standard",
                "imageDetails": {
                    "id": 123,
                    "versionId": 2,
                },
                "resources": {
                    "instanceTypeId": 21,
                    "storage": "15GB",
                },
                "command": "python main.py",
            },
        },
    },
    "release": {
        "version": "2.0.0",
        "notes": "This is a revised release",
    },
}

workflow_parameters: dict[str, Any] = {
    "run": [
        {
            "defaultValue": "AVG",
            "description": "Select an aggregation function (e.g., AVG, SUM, COUNT)",
            "name": "agg_type",
            "required": False,
            "title": "Agg Type",
            "type": "string",
        },
        {
            "defaultValue": False,
            "description": "Enable email notifications",
            "name": "email_notifications",
            "required": False,
            "title": "Email Notifications",
            "type": "boolean",
        },
        {
            "defaultValue": 10,
            "description": "Select the number of iterations",
            "name": "num_iterations",
            "options": [
                {
                    "title": "Low",
                    "value": 10,
                },
                {
                    "title": "Medium",
                    "value": 50,
                },
                {
                    "title": "High",
                    "value": 100,
                },
            ],
            "required": False,
            "title": "Number of Iterations",
            "type": "number",
        },
        {
            "defaultValue": ["input.csv", "output.csv"],
            "description": "Specify input and output file names",
            "name": "file_names",
            "required": True,
            "title": "File Names",
            "type": "string_array",
        },
    ],
}


new_spec_release: dict[str, str] = client.create_spec_release(
    spec_id=workflow_private_spec["id"],
    body=body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    parameters=workflow_parameters,
)

Create a new webapp block-spec release with updated config and release

body: dict[str, Any] = {
    "config": {
        "image": {
            "buildArguments": {
                "CHUNK_SIZE": "100",
            },
            "context": ".",
            "dockerfile": "Dockerfile",
            "version": "0.0.2",
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "sessionStickiness": True,
    },
    "release": {
        "version": "2.0.0",
        "notes": "This is a revised release",
    },
}

webapp_parameters: dict[str, Any] = {
    "run": [
        {
            "defaultValue": "AVG",
            "description": "Select an aggregation function (e.g., AVG, SUM, COUNT)",
            "name": "agg_type",
            "required": False,
            "title": "Agg Type",
            "type": "string",
        },
        {
            "defaultValue": False,
            "description": "Enable email notifications",
            "name": "email_notifications",
            "required": False,
            "title": "Email Notifications",
            "type": "boolean",
        },
        {
            "defaultValue": 10,
            "description": "Select the number of iterations",
            "name": "num_iterations",
            "options": [
                {
                    "title": "Low",
                    "value": 10,
                },
                {
                    "title": "Medium",
                    "value": 50,
                },
                {
                    "title": "High",
                    "value": 100,
                },
            ],
            "required": False,
            "title": "Number of Iterations",
            "type": "number",
        },
        {
            "defaultValue": ["input.csv", "output.csv"],
            "description": "Specify input and output file names",
            "name": "file_names",
            "required": True,
            "title": "File Names",
            "type": "string_array",
        },
    ],
}

new_spec_release: dict[str, str] = client.create_spec_release(
    spec_id=webapp_private_spec["id"],
    body=body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    parameters=webapp_parameters,
)

Describe the block-spec release

block_spec_release_info: dict[str, Any] = client.describe_spec_release(workflow_private_spec["id"], "2.0.0")

List all release(s) in a block-spec and iterate over them

spec_releases_iterator: Iterator[dict[str, Any]] = client.list_spec_releases(workflow_private_spec["id"])
spec_releases_iterated: dict[str, Any] = next(spec_releases_iterator)

Create a new deployment from a specific block-spec’s release

body: dict[str, str] = {
    "metadata": {
        "description": "Creating a new workflow block deployment.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "name": "new-deployment",
        "summary": "Creating new deployment.",
        "tags": [
            {
                "name": "sdk",
            },
        ],
        "title": "New Deployment",
    },
    "parameters": {
        "run": {
            "agg_type": "AVG",
            "email_notifications": True,
            "num_iterations": 50,
            "file_names": ["input.csv", "output.csv"],
        },
    },
    "revision": {
        "notes": "This is the initial revision",
    },
    "spec": {
        "id": workflow_private_spec["id"],
        "release": {
            "version": "1.0.0",
        },
    },
}

workflow_block_deployment: dict[str, str] = client.create_deployment(body)

List all existing deployment(s) and iterate over them

list_deployments_iterator: Iterator[dict[str, Any]] = client.list_deployments()
list_deployments_iterated: dict[str, Any] = next(list_deployments_iterator)

Describe a block deployment

client.describe_deployment(workflow_block_deployment["id"])

Update an existing deployment’s metadata

updated_deployment_body = {
    "description": "Updating workflow deployment",
    "descriptionContentType": "text/markdown",
    "imageUrl": "https://my-spec-pics.com/image-1.jpg",
    "name": "updated-deployment",
    "summary": "Updating the deployment metadata",
    "tags": [
        {
            "name": "sdk",
        },
    ],
    "title": "Updated Deployment",
}

client.update_deployment_metadata(workflow_block_deployment["id"], body=updated_deployment_body)

Get block deployment parameters

By default, it automatically picks the Deployment ID from the PRESS_DEPLOYMENT_ID environment variable. So, just running the following statement gives us all the parameters.

client.get_parameters()

The PRESS_DEPLOYMENT_ID environment variable is automatically added to Workflows and Webapps, so we can get the parameters by just running the above command.

If the environment variable is not present, we can pass in the value of the Deployment ID to the command as well.

client.get_parameters(deployment_id=workflow_block_deployment["id"])

Environment variable is given the highest priority. So, if the environment variable is present and we pass the deployment id as well, the fallback_params_file in environment variable will be used.

During development, you might not have the Deployment id. To be able to use the get-parameters function in that case, you can use a params file. For this purpose, just pass the path to a valid JSON file to the fallback_params_file argument of the function.

client.get_parameters(fallback_params_file="fallback.yaml")

Update block deployment parameters

parameters = {
    "agg_type": "MAX",
    "num_iterations": 10,
    "email_notifications": False,
    "file_names": ["input.csv", "output.csv"],
}

client.patch_parameters(deployment_id=workflow_block_deployment["id"], body=parameters)

Create block deployment revision from a specific block-spec’s release

body: dict[str, str] = {
    "parameters": {
        "run": {
            "agg_type": "AVG",
            "email_notifications": True,
            "num_iterations": 50,
            "file_names": ["input.csv", "output.csv"],
        },
    },
    "revision": {
        "notes": "This is the second revision",
    },
    "spec": {
        "id": workflow_private_spec["id"],
        "release": {
            "version": "2.0.0",
        },
    },
}


workflow_block_deployment_revision: dict[str, str] = client.create_deployment_revision(
    workflow_block_deployment["id"],
    body,
)

Describe a block deployment revision

client.describe_deployment_revision(workflow_block_deployment["id"], workflow_block_deployment_revision["revision"])

List all revision(s) in a block deployment and iterate over them

deployment_revisions_iterator: Iterator[dict[str, Any]] = client.list_deployment_revisions(
    workflow_block_deployment["id"],
)
deployment_revisions_iterated: dict[str, Any] = next(deployment_revisions_iterator)

Delete a block deployment

client.delete_deployment(workflow_block_deployment["id"])

Delete an existing block-spec with all it’s release(s)

client.delete_spec(workflow_private_spec["id"])