Devo SOAR - Integration Action

We have already discussed how to define an action. Let's enrich our action

import time
from lhub_integ.params import JinjaTemplatedStr
from lhub_integ import action

# ActionParam should be called with an ‘action’ argument to specify what action it is used for
PROJECT = ActionParam("PROJECT", description="Jira Project", action="process")

# accepts "process,process1" or ["process","process1"] or ("process","process1")
AP = ActionParam("SHARED_AP", description="Shared action param", action=["process", "process1"])


# A main function for the integraion must be defined. By default, we'll look for a function called 
# `process` in `main.py` but this can be overriden.

# Every method with `@action` will be converted into an action, adding (“...”) will define a name for the action 
@action("Process")
# adding a type int will cause the shim to coerce types properly
def process(summary, issue_type: int):
    """
    File a JIRA issue
    :param summary: Summary of the JIRA issue
    :label issue_type: Issue Type
    :optional summary: False
    :param issue_type: The type of JIRA issue
    :return:
    """
    # ^^^ An optional doc string may be provided in the standard Python docstring format.
    # If provided, it will be used to set the descriptions and labels for arguments in the integration


    # ... actually file a JIRA issue or something

    # The function must return either:
    # 1. A Python dictionary that is JSON serializable
    # 2. A list of Python dictionaries that are JSON serializable
    return {
        "status": "OK",
        "url": URL.read(),   # reading from the parameter
        "shared_action_param": AP.read()
    }

@action("Process1")
def process():
    return {
      "shared_action_param": AP.read()
    }

Important points to note

Here, process action only supports 2 parameters.
summary - Its type is not defined; by default, it will be a string.
issue_type - It will be attempted to be converted to an integer before passing it to the action function.
DocString - It supports providing information about function arguments.
label - Sets the label for the argument.
optional - Determines whether the argument is optional or not.
param - Sets the description of the field.

Note

Actions doesn't support vargs, kwargs right now i.e. you will get error if try to take variable number of arguments as input to the action function.

Action Validator

As discussed, we can verify Action Parameters before running the action.

import time
from lhub_integ.params import JinjaTemplatedStr
from lhub_integ import action
from lhub_integ.params import ConnectionParam, ActionParam, DataType, InputType, ValidationError
from typing import List, Optional


SLACK_INCOMING_WEBHOOK = ConnectionParam(
    id="slack_incoming_webhook",
    label="Incoming Webhook URL",
    description="Incoming Webhook URL for Slack",
    optional=False,
    options=None,
    data_type=DataType.STRING,
    input_type=InputType.PASSWORD
)

SLACK_CHANNEL_NAME = ActionParam(
    id="slack_channel_name",
    label="Channel Name",
    description="Jinja Templated Incoming Webhook has a default channel, but it can be overridden e.g. #general, #dev etc.",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.JINJA,
    input_type=InputType.TEXTAREA
)
SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS = ActionParam(
    id="slack_time_between_consecutive_requests_milliseconds",
    label="Time between consecutive API requests (in millis)",
    description="Time to wait between consecutive API requests in milliseconds (Default is 0 millisecond)",
    action=['process_post_message'],
    optional=True,
    options=None,
    data_type=DataType.INT,
    input_type=InputType.TEXT,
    default="0"
)

def validate_post_message() -&gt; Optional[List[ValidationError]]:
    ERRORS = []
    if len(SLACK_CHANNEL_NAME.read()) &gt; 50:
      ERRORS.append(ValidationError("Channel can't be more than 50 character long", SLACK_CHANNEL_NAME))
    return ERRORS


@action(name="Post Message", validator=validate_post_message)
def process_post_message(message: JinjaTemplatedStr):
    """
    Post a message on Slack

    :param message: Jinja-templated message string that will be posted on slack. Eg: '{{message_col}}'
    :label message: Message Template
    :optional message: False
    """
    post_message_on_slack(incoming_webhook=SLACK_INCOMING_WEBHOOK.read(), message=message, channel_name=SLACK_CHANNEL_NAME.read())

    time_delay = helpers.convert_milliseconds_to_seconds(SLACK_TIME_BETWEEN_CONSECUTIVE_REQUESTS_MILLISECONDS.read())
    time.sleep(time_delay)
    return {
        "result": "Successfully posted message to Slack"
    }