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()

Possible field values

Block kind

The kind of the block. The following kinds are supported: workflow and service.

Scope

The scope of the block. The following scopes are supported: private, public, and shared.

Spec status

The status of the block specs. The following statuses are supported: available, unavailable, and archived.

Deployment status

The status of the block deployments. The following statuses are supported: deployed, deploying, failed, redeploying, deleting, delete_failed, platform_resource_error, rollback, rollback_complete, rollback_failed and warning.

Service Type

The type of the service created from service kind of block. The following types are supported: api and web-app.

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,
                },
            },
            {
                "email": {
                    "name": "email-watcher-1",
                    "recipients": {
                        "to": ["user1@peak.ai", "user2@peak.ai"],
                    },
                },
                "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,
    auto_run_on_deploy=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,
                },
            },
            {
                "email": {
                    "name": "email-watcher-1",
                    "recipients": {
                        "to": ["user1@peak.ai", "user2@peak.ai"],
                    },
                },
                "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,
    auto_run_on_deploy=True,
    scope="shared",
    tenants=["tenant1", "tenant2"],
)

Prepare payload and create a new service `block-spec`

[ ]:

service_private_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "service",
    "metadata": {
        "name": "service-block-private",
        "title": "Web App type service block",
        "summary": "Creating a new service block spec.",
        "description": "Creating a new service 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": {
        "serviceType": "web-app",
        "imageDetails": {
            "id": 123,
            "versionId": 1,
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "sessionStickiness": True,
        "parameters": {
            "env": {
                "PARAM_1": "value1",
                "PARAM_2": "value2",
            },
            "secrets": ["SECRET_1", "SECRET_2"],
        },
        "entrypoint": "python main.py",
        "healthCheckURL": "/health",
        "minInstances": 1,
    },
}

service_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",
        },
    ],
}

service_private_spec: dict[str, str] = client.create_spec(
    body=service_private_spec_body,
    featured=True,
    scope="private",
    parameters=service_parameters,
)

service_shared_spec_body: dict[str, Any] = {
    "version": "1",
    "kind": "service",
    "metadata": {
        "name": "service-block-shared",
        "title": "API type service block",
        "summary": "Creating a new service block spec.",
        "description": "Creating a new service 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": {
        "serviceType": "api",
        "image": {
            "buildArguments": {
                "CHUNK_SIZE": "1000",
            },
            "context": ".",
            "dockerfile": "Dockerfile",
            "version": "0.0.1",
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "parameters": {
            "env": {
                "PARAM_1": "value1",
                "PARAM_2": "value2",
            },
            "secrets": ["SECRET_1", "SECRET_2"],
        },
        "entrypoint": "python main.py",
        "healthCheckURL": "/health",
        "minInstances": 1,
    },
}

service_shared_spec: dict[str, str] = client.create_spec(
    body=service_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)

"""
The list of block specs will be returned in the following format:

{
  "pageCount": 1,
  "pageNumber": 1,
  "pageSize": 25,
  "specCount": 1,
  "specs": [
    {
      "featured": false,
      "id": "0bddb4c6-40c5-45c3-b477-fceb2c051609",
      "kind": "workflow",
      "latestRelease": {
        "createdAt": "2020-01-01T18:00:00.000Z",
        "createdBy": "jane.smith@peak.ai",
        "id": "df113d64-ff44-4aa0-9278-edb03dae7a3f",
        "notes": "This is the original release",
        "version": "1.0.0",
        "autoRunOnDeploy": true
      },
      "metadata": {
        "createdAt": "2020-01-01T18:00:00.000Z",
        "createdBy": "jane.smith@peak.ai",
        "description": "This Workflow loads Opportunities and Accounts table from Salesforce into Peak-managed Snowflake.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "name": "load-salesforce-objects",
        "status": "available",
        "summary": "Load Opportunities and Accounts table from Salesforce.",
        "tags": [
          {
            "name": "salesforce"
          }
        ],
        "title": "Load Salesforce Objects",
        "updatedAt": "2020-01-01T18:00:00.000Z",
        "updatedBy": "jane.smith@peak.ai"
      },
      "scope": "private",
      "tenants": []
    }
  ]
}
"""

Describe an already existing `block-spec`

[ ]:

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

"""
The block spec details will be returned in the following format:

{
  "featured": false,
  "id": "0bddb4c6-40c5-45c3-b477-fceb2c051609",
  "kind": "workflow",
  "latestRelease": {
    "artifact": {
      "id": "721d738a-29f3-43b2-af52-c9055abe60b6",
      "version": 1
    },
    "config": {
      "images": {...},
      "steps": {...},
      "triggers": [...],
      "watchers": [...]
    },
    "createdAt": "2020-01-01T18:00:00.000Z",
    "createdBy": "jane.smith@peak.ai",
    "id": "df113d64-ff44-4aa0-9278-edb03dae7a3f",
    "notes": "This is the original release",
    "version": "1.0.0",
    "autoRunOnDeploy": true
  },
  "metadata": {
    "createdAt": "2020-01-01T18:00:00.000Z",
    "createdBy": "jane.smith@peak.ai",
    "description": "This Workflow loads Opportunities and Accounts table from Salesforce into Peak-managed Snowflake.",
    "descriptionContentType": "text/markdown",
    "imageUrl": "https://my-block-pics.com/image-0.jpg",
    "name": "load-salesforce-objects",
    "status": "available",
    "summary": "Load Opportunities and Accounts table from Salesforce.",
    "tags": [
      {
        "name": "salesforce"
      }
    ],
    "title": "Load Salesforce Objects",
    "updatedAt": "2020-01-01T18:00:00.000Z",
    "updatedBy": "jane.smith@peak.ai"
  },
  "parameters": {
    "build": [...],
    "groups": {
      "build": [...],
      "run": [...]
    },
    "run": [...]
  },
  "scope": "private",
  "tenants": []
}
"""

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,
                },
            },
            {
                "email": {
                    "name": "email-watcher-1",
                    "recipients": {
                        "to": ["user1@peak.ai", "user2@peak.ai"],
                    },
                },
                "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,
    auto_run_on_deploy=True,
)

Create a new service `block-spec` release with updated config and release

[ ]:

body: dict[str, Any] = {
    "config": {
        "serviceType": "web-app",
        "image": {
            "buildArguments": {
                "CHUNK_SIZE": "100",
            },
            "context": ".",
            "dockerfile": "Dockerfile",
            "version": "0.0.2",
        },
        "resources": {
            "instanceTypeId": 1,
        },
        "sessionStickiness": False,
        "parameters": {
            "env": {
                "PARAM_1": "value1",
                "PARAM_2": "value2",
            },
            "secrets": ["SECRET_1", "SECRET_2"],
        },
        "entrypoint": "python main.py",
        "healthCheckURL": "/health",
        "minInstances": 1,
    },
    "release": {
        "version": "2.0.0",
        "notes": "This is a revised release",
    },
}

service_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="<spec_id>",
    body=body,
    artifact={"path": "../peak", "ignore_files": [".dockerignore"]},
    parameters=service_parameters,
)

Describe the `block-spec` release

[ ]:

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

"""
The block spec release details will be returned in the following format:

{
  "artifact": {
    "id": "721d738a-29f3-43b2-af52-c9055abe60b6",
    "version": 1
  },
  "config": {
    "images": {...},
    "steps": {...},
    "triggers": [...],
    "watchers": [...]
  },
  "createdAt": "2020-01-01T18:00:00.000Z",
  "createdBy": "jane.smith@peak.ai",
  "id": "df113d64-ff44-4aa0-9278-edb03dae7a3f",
  "notes": "This is the original release",
  "autoRunOnDeploy": true,
  "parameters": {
    "build": [...],
    "groups": {
      "build": [...],
      "run": [...]
    },
    "run": [...]
  }
}
"""

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)

"""
The list of block spec releases will be returned in the following format:

{
  "pageCount": 1,
  "pageNumber": 1,
  "pageSize": 25,
  "releaseCount": 1,
  "releases": [
    {
      "createdAt": "2020-01-01T18:00:00.000Z",
      "createdBy": "jane.smith@peak.ai",
      "id": "df113d64-ff44-4aa0-9278-edb03dae7a3f",
      "notes": "This is the original release",
      "version": "1.0.0",
      "autoRunOnDeploy": true
    }
  ]
}
"""

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",  # Release version of the parent spec to be used
        },
    },
}

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)

"""
The list of block deployments will be returned in the following format:

{
  "deploymentCount": 1,
  "deployments": [
    {
      "id": "e04e308d-98dc-4443-bb02-c4ef990a3fdc",
      "kind": "workflow",
      "latestRevision": {
        "createdAt": "2020-01-01T18:00:00.000Z",
        "createdBy": "jane.smith@peak.ai",
        "id": "7092bd84-c35d-43c1-90ca-7510a1204dcc",
        "notes": "This is the initial revision",
        "revision": 1,
        "status": "deploying"
      },
      "metadata": {
        "createdAt": "2020-01-01T18:00:00.000Z",
        "createdBy": "jane.smith@peak.ai",
        "description": "This Workflow loads Opportunities and Accounts table from Salesforce into Peak-managed Snowflake.",
        "descriptionContentType": "text/markdown",
        "imageUrl": "https://my-block-pics.com/image-0.jpg",
        "name": "lso-deployment",
        "status": "deploying",
        "summary": "Load Opportunities and Accounts table from Salesforce.",
        "tags": [
          {
            "name": "salesforce"
          }
        ],
        "title": "LSO Deployment",
        "updatedAt": "2020-01-01T18:00:00.000Z",
        "updatedBy": "jane.smith@peak.ai"
      },
      "spec": {
        "id": "0bddb4c6-40c5-45c3-b477-fceb2c051609",
        "release": {
          "currentVersion": "1.0.0",
          "latestVersion": "1.0.0"
        }
      }
    }
  ],
  "pageCount": 1,
  "pageNumber": 1,
  "pageSize": 25
}
"""

Describe a `block` `deployment`

[ ]:

client.describe_deployment(workflow_block_deployment["id"])

"""
The block deployment details will be returned in the following format:

{
  "id": "e04e308d-98dc-4443-bb02-c4ef990a3fdc",
  "kind": "workflow",
  "latestRevision": {
    "resources": [
      {
        "errors": [
          {
            "code": 500,
            "createdAt": "2020-01-01T18:00:00.000Z",
            "message": "This resource failed to deploy"
          }
        ],
        "id": "9e1389f2-564c-4d60-80cf-5369d8a5a204",
        "kind": "workflow",
        "name": "load-salesforce-objects",
        "status": "deploying",
        "version": "0.0.1",
        "versionId": "1"
      }
    ],
    "createdAt": "2020-01-01T18:00:00.000Z",
    "createdBy": "jane.smith@peak.ai",
    "id": "7092bd84-c35d-43c1-90ca-7510a1204dcc",
    "notes": "This is the initial revision",
    "revision": 1,
    "status": "deploying"
  },
  "metadata": {
    "createdAt": "2020-01-01T18:00:00.000Z",
    "createdBy": "jane.smith@peak.ai",
    "description": "This Workflow loads Opportunities and Accounts table from Salesforce into Peak-managed Snowflake.",
    "descriptionContentType": "text/markdown",
    "imageUrl": "https://my-block-pics.com/image-0.jpg",
    "name": "lso-deployment",
    "status": "deploying",
    "summary": "Load Opportunities and Accounts table from Salesforce.",
    "tags": [
      {
        "name": "salesforce"
      }
    ],
    "title": "LSO Deployment",
    "updatedAt": "2020-01-01T18:00:00.000Z",
    "updatedBy": "jane.smith@peak.ai"
  },
  "parameters": {
    "run": [...]
  },
  "parent": {
    "id": "a3e77006-86f3-4829-8c43-f21ad462dbbd"
  },
  "spec": {
    "id": "0bddb4c6-40c5-45c3-b477-fceb2c051609",
    "release": {
      "currentVersion": "1.0.0",
      "latestVersion": "1.0.0"
    }
  }
}
"""

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 Services (Web Apps and APIs), 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",
    },
    "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"])

"""
The block deployment revision details will be returned in the following format:

{
  "resources": [
    {
      "errors": [
        {
          "code": 500,
          "createdAt": "2020-01-01T18:00:00.000Z",
          "message": "This resource failed to deploy"
        }
      ],
      "id": "9e1389f2-564c-4d60-80cf-5369d8a5a204",
      "kind": "workflow",
      "name": "load-salesforce-objects",
      "status": "deploying",
      "version": "0.0.1",
      "versionId": "1"
    }
  ],
  "createdAt": "2020-01-01T18:00:00.000Z",
  "createdBy": "jane.smith@peak.ai",
  "id": "7092bd84-c35d-43c1-90ca-7510a1204dcc",
  "notes": "This is the initial revision",
  "revision": 1,
  "status": "deploying",
  "spec": {
    "id": "0bddb4c6-40c5-45c3-b477-fceb2c051609",
    "release": {
      "version": "1.0.0"
    }
  },
  "parameters": {
    "run": [...]
  }
}
"""

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)

"""
The list of block deployment revisions will be returned in the following format:

{
  "pageCount": 1,
  "pageNumber": 1,
  "pageSize": 25,
  "revisionCount": 1,
  "revisions": [
    {
      "createdAt": "2020-01-01T18:00:00.000Z",
      "createdBy": "jane.smith@peak.ai",
      "id": "7092bd84-c35d-43c1-90ca-7510a1204dcc",
      "notes": "This is the initial revision",
      "revision": 1,
      "status": "deploying"
    }
  ]
}
"""

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"])

Press Blocks

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

Possible field values

Block kind

Scope

Spec status

Deployment status

Service Type

Prepare payload and create a new workflow block-spec

Prepare payload and create a new service block-spec

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

Describe an already existing block-spec

Update the metadata and discoverability of a block-spec

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

Create a new service block-spec release with updated config and release

Describe the block-spec release

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

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

List all existing deployment(s) and iterate over them

Describe a block deployment

Update an existing deployment’s metadata

Get block deployment parameters

Update block deployment parameters

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

Describe a block deployment revision

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

Delete a block deployment

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