Python operations

API Object #

Run #

api.run(operation, parameters, options)

Runs an operation.

ArgumentTypeDescription
operationStringThe name of the operation to be run
parametersDictionary(Optional) an object containing any operation-defined parameters
optionsDictionary(Optional) additional options:limit - limits the number of results returned. Unbounded if undefined.

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.

ArgumentTypeDescription
operationsList of dictionariesEach dictionary describes one of the operations to run in parallel (See below)

Each dictionary in operations has the following properties:

FieldTypeDescription
operationStringThe name of the operation to be run
parametersDictionary(Optional) an object containing any operation-defined parameters
optionsDictionary(Optional) additional options - limit: limits the number of results returned, infinity if undefined.

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.

ArgumentTypeDescription
queryStringThe SQL query to be run
parametersDictionary(Optional) an object containing any operation-defined parameters

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.

ArgumentTypeDescription
slackList of DictionariesSlack Block Kit blocks, see how to form the blocks here
metadataDictionaryActs as a data blob for future use, send as empty dictionary for now (Example: {})
sensitiveBoolean(Optional) Defaults to false. If true, the blocks will be posted ephemerally to Slack, and the timeline will display "sensitive message" in place of the blocks.

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.
ArgumentTypeDescription
slackList of DictionariesSlack Block Kit blocks, see how to form the blocks here
metadataDictionaryActs as a data blob for future use, send as empty dictionary for now (Example: {})
sensitiveBoolean(Optional) Defaults to false. If true, the blocks will be posted ephemerally to Slack, and the timeline will display "sensitive message" in place of the blocks.

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.
ArgumentTypeDescription
messageStringA plain text, failure message for Slack and the timeline
metadataDictionaryActs as a data blob for future use, send as empty dictionary for now (Example: {})
sensitiveBoolean(Optional) Defaults to false. If true, the blocks will be posted ephemerally to Slack, and the timeline will display "sensitive message" in place of the blocks.

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

ArgumentTypeDescription
messageStringA plain text, warning message for Slack and the timeline
metadataDictionaryActs as a data blob for future use, send as empty dictionary for now (Example: {})
sensitiveBoolean(Optional) Defaults to false. If true, the blocks will be posted ephemerally to Slack, and the timeline will display "sensitive message" in place of the blocks.

Returns (Null)

Example

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

Status #

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

Log a status update.

ArgumentTypeDescription
messageStringA plain text, status update message for Slack and the timeline
metadataDictionaryActs as a data blob for future use, send as empty dictionary for now (Example: {})
sensitiveBoolean(Optional) Defaults to false. If true, the blocks will be posted ephemerally to Slack, and the timeline will display "sensitive message" in place of the blocks.
groupIdStringAny arbitrary message or id string, used to group together status updates from a specific workflow

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.

ArgumentTypeDescription
alertInformationDictionaryAn dictionary containing optional information about the alert (See below)
blocksList of DictionariesSlack Block Kit blocks, see how to form the blocks here

Each alertInformation dictionary has the following properties:

FieldTypeDescription
nameString(Optional) The alert name
externalUrlString(Optional) a 3rd-party link to the alert or issue
metadataDictionary(Optional) additional information, like the 3rd-party alert ID

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.

ArgumentType
keyStringthe key of the 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 all of the Python Standard Library, as well as popular Python packages such as pytz. Below are some examples of useful modules:

import json
data = {"hello": "world"}
print(json.dumps(data))
// => "{"hello": "world"}"
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"
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="
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.