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,
"webhookPolicy": "generate",
},
],
"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_IDenvironment 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"])