Python operations

API Object #

Run #

api.run(operation, parameters, options)

Runs an operation.

Note: If you are passing in operation and options, but no parameters, you will need to pass in an empty parameters dictionary.

Returns (List): Returns the operation results or throws any operation failure.

Examples

api.run("this.hello_world")
// => [{"Hello": "World"}]

api.run("this.restart_vm", { "id": params.userId })
//  passing a parameter to an operation
// => [{id:1234, ...}]

api.run("this.metrics", {}, {"limit": 10})
// => returns result array of size 10

Run bulk #

api.runBulk(operations)

Runs any number of operations in parallel and waits for all to complete, or any to fail, before returning.

Each dictionary in operations has the following properties:

Returns (List): Returns a list where each element is the result from one of the parallel operations. If any fail, an error is thrown.

Example

results = api.runBulk([{
  "operation": "this.hello_world"
}, {
  "operation": "this.restart_vm",
  "parameters": { "id": params.userId }
}, {
  "operation": "this.metrics",
  "options": { "limit": 10 }
}])

results[0] // => [{"Hello": "World"}]
results[1] // => [{id:1234, ...}]
results[2] // => result list of size 10

Query #

api.query(query, parameters)

Run a SQL query. Note that if you have dynamic fields in your query, using parameters is highly recommended to avoid escaping issues and SQL injection attacks.

Returns (List): Returns the query results or throws any query failure.

Example

api.query("select * from this.hello_world")
// => [{"Hello": "World"}]

api.query("select * from this.restart_vm where id=@userId", {"userId": params.userId});
// => [{id:1234, ...}]

Workflows #

Important: All workflow operations require a context parameter in the operation they are running in. This ensures the workflow operations are posting in the correct Slack channel and have other required contextual data.

Info #

workflow.log.info(slack=[blocks], metadata = {}, sensitive)

Log anything.

This binding is only intended for notifications that don’t fit the other workflow.log bindings. Before using this binding, first try to fit your message into any of the other bindings. For example, workflow results and data should go with the done binding while status updates should use the status binding.

Returns (Null)

Example

output_blocks = api.run("block_kit_lib.markdown_text_section", {"text": f"Hello World"})
workflow.log.info(slack=output_blocks, metadata={})

Done #

workflow.log.done(slack, metadata, sensitive)

Log results and mark the successful completion of an entire workflow run.

Once and only once: This binding should be used once and only once in a single workflow run:

• At least once: It is highly encouraged, but not technically enforced, that done is used at least once. The only reason it is not enforced is for backwards compatibility. We may enforce this in the future!
• Not more than once: A workflow will throw an error if it is used more than once. This is because the done binding is intended to mark the true end of a workflow run.
• Oneof done or fail: The workflow will also fail if done is called after fail because a workflow cannot both finish and fail.

Returns (Null)

Example

output_blocks = api.run("block_kit_lib.markdown_text_section", {"text": f"Hello World"})
workflow.log.done(slack=output_blocks, metadata={})

Fail #

workflow.log.fail(message, metadata={}, sensitive)

Log a failure of and mark the completion of an entire workflow run as a failure.

Once and only once: This binding should be used once and only once in a single workflow run:

• Not more than once: A workflow will throw an error if the fail binding is used more than once in a single run. This is because the fail binding is intended for true failures of the workflow itself. If you find yourself wanting to write multiple fails for a single code path, you should instead use warn.
• Oneof done or fail: The workflow will also fail if fail is called after done because a workflow cannot both finish and fail.

Returns (Null)

Example

workflow.log.fail(message="Restart VM workflow failed", metadata={})

Warn #

workflow.log.warn(message, metadata={}, sensitive)

Log a warning message. This includes errors that do not cause the entire workflow to fail (see workflow.log.fail).

Returns (Null)

Example

workflow.log.warn(message="arbitrary message", metadata={}, sensitive=True)

Status #

workflow.log.status(message, metadata, sensitive, groupId)

Log a status update.

Returns (Null)

Example

workflow.log.status(message="status update message", metadata={}, sensitive=False, groupId="restart_vm")

Integrators #

integrator.response(alertInformation, blocks)

Posts a message to the connected trigger channel about the received alert. The message posted will also contain a button to create an Incident.

Each alertInformation dictionary has the following properties:

Returns (Null)

Example

output_blocks = api.run("block_kit_lib.markdown_text_section", {"text": f"Hello World"})
integrator.response({"name": "Test integrator alert"}, output_blocks)

Environment Variables #

Transposit operations can programmatically access environment variables set in Workflow configuration.

get #

env.get(key)

Retrieves the value set for the specified environment variable.

Returns (String/Number/Boolean/Null): Returns a value matching the value that was set in the UI. If the key does not match a value defined in the schema, an exception is thrown. If no saved value exists, and no default value is defined, returns Null.

Examples

env.get("stringKey")
// => "stringValue"
env.get("numberKey")
// => 88888
env.get("booleanKey")
// => True
env.get("unsetKey")
// => null
env.get("undefinedKey")
// => exception

getBuiltin #

env.getBuiltin()

Returns (Dictionary): Returns a Dictionary containing builtin environment variables.

Examples

env.getBuiltin()
// => {
//      "appUrl": "https://example-hd39c.transposit.io",
//      "requestId": "bcbfa9b1-f9a9-45c7-aeaa-6e903153cd9e"
//    }

Available Python packages #

Transposit provides you access to a few different popular Python packages.

• json

import json
data = {"hello": "world"}
print(json.dumps(data))
// => "{"hello": "world"}"

• pytz, datetime

import pytz, datetime
us_eastern_tz = pytz.timezone('US/Eastern')
dt = us_eastern_tz.localize(datetime.datetime(2020, 11, 2, 15, 25, 0))
print(dt)
// => "2020-11-02 15:25:0-05:00"

• base64

import base64
output_text = "Hello World"
output_text_bytes = output_text.encode("utf-8")
print(output_text_bytes)
// =>  "b'Hello World'"

base64_bytes = base64.b64encode(output_text_bytes)
print(base64_bytes)
// => "b'SGVsbG8gV29ybGQ='"

base64_output_text = base64_bytes.decode("utf-8")
print(base64_output_text)
// => "SGVsbG8gV29ybGQ="

• re

import re
input = "abc123abc"
api.log(re.sub("abc", "def", input))
// => "def123def"

If you need a different Python package for your application to work, please reach out to use at support@transposit.com and let us know what you are looking for.