Invoke an action from a Python action script

Is there any way to invoke execution of an action from the python client? I would like to execute an action from an action without using action chain or mistral. I ended up using my one ActionManager but I wonder if there is a better API supported by the python client.

from st2client.client import Client
from st2client.models import LiveAction


class ActionManager(object):

    def __init__(self):
        self.client = Client(base_url='http://localhost')

    def run_action_by_name(self, name, payload):
        action = self._get_action_from_name(name)
        action_ref = '.'.join([action.pack, action.name])
        execution = LiveAction()
        execution.action = action_ref
		execution.parameters = payload
        action_exec_manager = self.client.managers['LiveAction']
        action_exec_manager.create(execution)

    def _get_action_from_name(self, name):
        return self.client.actions.get_by_name(name)

@n_kos Here’s a working code snippet to invoke an action and wait for the results. Hope this helps:

#!/usr/bin/env python
#
########################
# Install Instructions #
########################
#
# virtualenv virtualenv
# source ./virtualenv/bin/activate
# pip install st2client
# ./st2_action_exec.py

import st2client
import st2client.client
import st2client.commands.action
import st2client.models
import time
import sys

import requests
requests.packages.urllib3.disable_warnings()


PENDING_STATUSES = [
    st2client.commands.action.LIVEACTION_STATUS_REQUESTED,
    st2client.commands.action.LIVEACTION_STATUS_SCHEDULED,
    st2client.commands.action.LIVEACTION_STATUS_RUNNING,
    st2client.commands.action.LIVEACTION_STATUS_CANCELING
]

def run_action(server, username, password, ref, params):
    client = st2client.client.Client(base_url="https://{}/".format(server),
                                     auth_url="https://{}/auth/v1".format(server),
                                     api_url="https://{}/api/v1".format(server))

    # login
    token_instance = st2client.models.Token()
    token = client.tokens.create(token_instance, auth=(username, password))
    client = st2client.client.Client(base_url="https://{}/".format(server),
                                     auth_url="https://{}/auth/v1".format(server),
                                     api_url="https://{}/api/v1".format(server),
                                     token=token.token)

    # execute
    execution_instance = st2client.models.LiveAction()
    execution_instance.action = ref
    execution_instance.parameters = params
    execution = client.liveactions.create(execution_instance)

    # wait for the execution to finish
    while execution.status in PENDING_STATUSES:
        time.sleep(1)
        sys.stdout.write('.')
        sys.stdout.flush()
        print "waiting for execution {}".format(execution.id)
        execution = client.liveactions.get_by_id(execution.id)

    # execution is done, print out the details
    print execution.__dict__

if __name__ == "__main__":
    server = "stackstorm.domain.tld"
    username = "st2admin"
    password = "xxx"
    ref = "core.local"
    params = {"cmd": "ls"}
    run_action(server, username, password, ref, params)

FYI, this code is pieced together from parts of the st2client package:

@nmaludy
Thank you for your reply! I guess your script is more generic for usage outside the stackstorm server.
What I want to achieve is invoking an action from an action without using action chain or a mistral flow.For my case, my script is more a fit as you can import ActionManager in every action you want.

What I was looking for is an API direct from the client. Something like —>

action = client.actions.get_by_name(name)
action.run(parameters=payload)

ps I don’t care about the result within the Action but is a nice to have.

@n_kos this is using an api “directly from the client” by calling execution = client.liveactions.create(execution_instance)

However, this function does not take arbitrary parameters like you want, instead it requires a LiveAction object to be passed in.

About the shortest i could condense it down into is the following. This assumes that you’re running inside a StackStorm action because it relies on the ST2_AUTH_TOKEN environment variable being set.

from st2common.runners.base_action import Action

from st2client.client import Client
from st2client.models import LiveAction

class ActionManager(object):

    def execute(self, server, action, params):
        client = Client(base_url="https://{}/".format(server))
        client.liveactions.create(LiveAction(action=action, parameters=params))


class ActionRunner(Action):

    def run(self):
        am = ActionManager()
        am.execute("127.0.0.1",
                   "core.local",
                   {"cmd": "ls"})

I like it! Looks better than mine! Thank you

@nmaludy This is working as expected but action is getting triggered Asynchronously.
Is it possible to retrieve the result of action triggered in python script ?

@sakshi95gupta Yes, please see the first code snipped in my original post. Hint, it’s the code snipped with the while loop: while execution.status in PENDING_STATUSES:

Thanks a lot. It worked.