Automation-as-Code

Automation-as-Code is a way to create automations using Python code. Benefits include:

Being able to leverage software engineering best practices:
- Reduction of errors through automated testing when deploying an automation
- Version Control
- Pull requests (propose changes for review, encourage collaboration and discussion among team members, ensure good documentation)
Reduction in vendor lock-in as the automations are stored in plain Python code
Enabling to create custom actions, custom integrations with the flexibility of code
Easier reusability through a modularized setup of automations

Project Structure

For compatability and increased reusability, the recommended project setup in Git should be as follows:

- __init__.py
- custom_action1.py
- __init__.py
- workflow1.py

For an example, check out our Github example project (opens in a new tab).

The Workflow Function

The workflow function is a Python function that is used to specify the control flow of the actions and orchestrates the input and output for the actions. Think of it like the recipe for the ingredients (pre-build actions, integrations, and custom actions). The workflow function is more restrictive than pure Python code as Admyral needs to compile it and map it onto the No-Code interface.

A workflow function is defined as a Python function with the @workflow decorator:

from admyral.workflow import workflow
from adymral.typings import JsonValue
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    # some actions / control flow of your automation

Additionally, a workflow function must always have the parameter payload: dict[str, JsonValue]. No other parameters for the workflow function are allowed.

Workflow Decorator

A workflow function must always be decorated with the @workflow decorator imported via from admyral.workflow import workflow. The workflow decorator supports the following arguments:

description: Add a description to your workflow which will be displayed in the Admyral UI
triggers: Define triggers for your workflow. See the Triggers section below for more information.

Example:

from admyral.workflow import workflow, Webhook, Schedule
from adymral.typings import JsonValue
 
@workflow(
    description="This is my workflow description example",
    triggers=[
        Webhook(),
        Schedule(interval_hours=1)
    ]
)
def example_workflow(payload: dict[str, JsonValue]):
    # some actions / control flow of your automation

Triggers

Admyral supports the following trigger types:

Webhook: An event-based trigger which let's you use the workflow as an API.

In order to call your webhook, you need the webhook URL and the webhook secret. You can find both if you go to your workflow in the Admyral UI and click on the Start Worfklow node.

Option 1: Authenticate via the API URL
```
curl -X POST webhook-url/webhook-secret
```
Option 2: Authenticate via the header
```
curl -X POST webhook-url -H 'Authorization: webhook-secret'
```
Schedule: A schedule-based trigger. The following schedule types are supported:
- interval_seconds: Run the workflow every X seconds
- interval_minutes: Run the workflow every X minutes
- interval_hours: Run the workflow every X hours
- interval_days: Run the workflow every X days
- cron: Run the workflow based on the defined cron expression

Example:

from admyral.workflow import workflow, Webhook, Schedule
from adymral.typings import JsonValue
 
@workflow(
    description="This is my workflow description example",
    triggers=[
        Webhook(),
        Schedule(interval_seconds=60),
        Schedule(interval_hours=1),
        Schedule(cron="0 0 * * *"),
    ]
)
def example_workflow(payload: dict[str, JsonValue]):
    # some actions / control flow of your automation

Triggers - Default Arguments

Triggers also support default arguments. Default arguments are passed to triggers via keyword arguments. If a workflow execution is started by a trigger, the default arguments are injected into the payload dictionary if the key does not yet exist in the payload. This implies that for a scheduled execution, default arguments are always injected while for a webhook-based or manually-triggered execution a value is only injected if the key is not present in the incoming payload.

Example:

from admyral.workflow import workflow, Webhook, Schedule
from adymral.typings import JsonValue
 
@workflow(
    description="This is my workflow description example",
    triggers=[
        Webhook(
            # Define your default arguments as keyword arguments here
            some_key="hello from webhook"
        ),
        Schedule(
            interval_seconds=60,
            # Define your default arguments as keyword arguments here
            some_key="hello from schedule"
        ),
    ]
)
def example_workflow(payload: dict[str, JsonValue]):
    # some actions / control flow of your automation

Action Calling and Data Flow

Actions are called like normal functions and they can return data which can be assigned to variables. Hence, in order to pass output data from one action to another action, you need to assign the action call to variable:

result = my_action_call()

The data is always passed using keyword arguments:

result = my_action_call()
my_other_action(
    arg=result
)

This creates a dependency between the two actions which is visualized accordingly in the No-Code Editor (my_action_call --> my_other_action). Due to the dependency, Admyral will always first execute my_action_call and then my_other_action because my_other_action depends on the output of my_action_call.

The following semantics are supported for passing data as arguments to an action:

f-strings f"this is a string with some value: {x}"
constants of type int, float, str, list, dict, None, bool
dict constructions
list constructions
variables incl. access paths for possibly nested lists and dicts, e.g., payload["some_list"][0]["some_key"]

Example:

from admyral.workflow import workflow
from adymral.typings import JsonValue
from admyral.actions import send_email, send_slack_message_to_user_by_email
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    sent_emails = send_email(
        # You can manually build lists using constants and variables
        recipients=["daniel@admyral.dev", payload["recipient"]],
        sender_name="Admyral",
        subject="Test",
        body="Test Body"
    )
 
    send_slack_message_to_user_by_email(
        # use a variable to pass data
        email=payload["recipient"],
        # use an f-string for building formatted strings
        text=f"Hey! You received the following emails: {sent_emails}",
    )

Supported Data types

Currently, Admyral only supports JSON-serializable types as input and output for actions, i.e., the following types are supported:

str
None
bool
int
float
list consisting of JSON-serializable types
dict consisting of JSON-serializable types
our custom type representing a JSON-serializable type: JsonValue (Usage: from admyral.typings import JsonValue)

Secrets

A lot of actions interact with other tools which require authentication. Hence, they need to be given access to secrets.

If an action requires access to a secret, it defines a secret placeholder. In order to use this action, you need to map a secret to the secret placeholder when calling the action.

Example: Calling send_slack_message_to_user_by_email

The action send_slack_message_to_user_by_email requires a secret to be mapped to SLACK_SECRET. We previously setup the Slack connection by following the guide from our integrations documentation and stored the Slack secret with the name my_stored_slack_secret.

Hence, we map our stored Slack secret my_stored_slack_secret to the secret placeholder to give the action access to Slack:

from admyral.workflow import workflow
from admyral.typings import JsonValue
from admyral.actions import send_slack_message_to_user_by_email
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    send_slack_message_to_user_by_email(
        email="daniel@admyral.dev",
        text="Hello from Admyral",
        # map the stored slack secret to the placeholder
        secrets={
            "SLACK_SECRET": "my_stored_slack_secret"
        }
    )

If you build your custom actions, you don't need to define it because the secrets argument is automatically available if you define secret placeholders.

See Secrets for more information.

Custom Python Code

Admyral supports two options for leveraging custom Python code in your workflows:

Custom Actions: Build your own custom action and make it available for your code and No-Code workflow

Example:

from admyral.workflow import workflow
from admyral.typings import JsonValue
 
# Import your custom action in your workflow file
from actions.my_custom_actions import my_custom_action
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    # Use your custom action like you would use pre-built actions!
    my_custom_action()

Python Sections: A marked section where you can write custom Python code directly in the workflow function. A Python section starts with the # {% custom %} marker and ends with # {% endcustom %} marker. The Python section is displayed as a Python script node in the No-Code Editor.

Example:

from admyral.workflow import workflow
from admyral.typings import JsonValue
 
@workflow
def example_workflow():
    # some actions
 
    # {% custom %}
    # add custom python code here
    users = [a, b, c]
    for user in users:
        # do something
    # {% endcustom %}
 
    # continue your workflow function

Python sections are not yet available but will be released in the upcoming weeks.

See Custom Actions to learn more about using custom actions in your workflows.

If-Conditions

Within the workflow function, it is allowed to use if-statements and nested if-statements. Admyral compiles these into an If-Condition node in the No-Code editor.

from admyral.workflow import workflow
from admyral.typings import JsonValue
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    # previous workflow logic with action A and B
    if result_from_action_a == true and result_from_action_b:
        # do something

See If Conditions in Pre-built Actions for more information.

For Loops

For loops are not yet available but will be released in the upcoming weeks.

Handling Side-Effects with run_after

The Admyral compiler automatically parallelizes actions which are independent from each other. Since the compiler parallelizes the action based on the data flow in the workflow function, there might be situations where there are hidden dependencies between two actions due to some side-effects. In order to enforce the correct execution order, you can use the run_after argument.

This workflow

from admyral.workflow import workflow
from admyral.typings import JsonValue
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    custom_action1()
    custom_action2()

would be transformed into a No-Code workflow where the start workflow has two children custom_action1 and custom_action2. Since these actions are assumed to be independent, they are executed in parallel.

To ensure that both actions are executed in sequential order, use run_after:

from admyral.workflow import workflow
from admyral.typings import JsonValue
 
@workflow
def example_workflow(payload: dict[str, JsonValue]):
    # do a variable assignment - even if the action does not return anything
    result = custom_action1()
    custom_action2(
        # enforce that custom_action2 is executed after
        # custom_action1 by passing result to the run_after
        # argument
        run_after=[result]
    )

Note that all actions (custom and pre-built) support run_after. If you build your custom actions, you don't need to define it because the run_after argument is automatically available.

Executing an Automation

Automation based on Admyral is written in Python. This allows you to run and test your automations locally as standard Python scripts.

Note that for this to work, the secrets (e.g. API keys) must be available as environment variables.

To execute an automation locally, simply call it like a normal Python function

from admyral.workflow import workflow
from adymral.typings import JsonValue
 
@workflow
def example_workflow(dict: [str, JsonValue]):
    # some actions / control flow of your automation
    ...
 
if __name__ == "__main__":
    # just call your workflow function like a normal Python function
    example_workflow({})
 
    # you can also pass input via the payload to your workflow
    example_workflow({
        "some_argument": ...
    })

and then simply execute it as a normal Python script:

python example_workflow.py

To execute an automation using Admyral's workflow engine, you first need to push the workflow (Note: by adding --activate your workflow is also immediately activated) and then you can simply trigger it using the CLI:

# Push your workflow to Admyral
admyral workflow push example_workflow -f example_workflow.py --activate
# Trigger your workflow
admyral workflow trigger example_workflow

See Admyral CLI to learn more information about the available CLI commands.

Introduction Admyral CLI