The MSactivator™ provides a support for Python SDK dedicated to developing automation workflows.
Overview
This SDK provides a set of functions to call the MSactivator™ REST API and automate action on the MSactivator™ such as create and activate managed entities, call microservice functions, call processes from other workflows…
The SDK API documentation is available online on your MSactivator™ instance at: https://localhost/msa_sdk
Contribution
The sources of the SDK is available on Github.
Code samples
Sample implementation of a SDK function.
This code sample is an example of the implementation of a Python SDK function to create a new managed entity.
class Device(MSA_API):
self.api_path = "/device"
def create(self):
self.action = 'Create device'
self.path = '{}/v2/{}'.format(self.api_path, self.customer_id) (1)
data = {"manufacturerId": self.manufacturer_id,
"modelId": self.model_id,
"managementAddress": self.management_address,
"reporting": self.reporting,
"useNat": self.use_nat,
"logEnabled": self.log_enabled,
"logMoreEnabled": self.log_more_enabled,
"managementInterface": self.management_interface,
"mailAlerting": self.mail_alerting,
"passwordAdmin": self.password_admin,
"externalReference": self.device_external,
"login": self.login,
"name": self.name,
"password": self.password,
"id": 0,
"snmpCommunity": self.snmp_community}
if self.management_port:
data["managementPort"] = self.management_port
self.call_post(data) (2)
self.fail = not self.response.ok
if self.response.ok:
self.device_id = json.loads(self.content)['id']
return json.loads(self.content)| 1 | the REST API to call |
| 2 | post the data to the REST API |
Sample call of a function in a workflow task.
from msa_sdk.device import Device
from msa_sdk.variables import Variables
import json
dev_var = Variables() (1)
dev_var.add('customer_id')
dev_var.add('managed_device_name')
dev_var.add('manufacturer_id')
dev_var.add('model_id')
dev_var.add('device_ip_address')
dev_var.add('login')
dev_var.add('password')
dev_var.add('password_admin')
context = Variables.task_call(dev_var)
new_device = Device(context['customer_id'], context['managed_device_name'], context['manufacturer_id'],context['model_id'], context['login'], context['password'], context['password_admin'],context['device_ip_address'])
new_device.create() (2)
context['device_id'] = new_device.device_id (3)
print(new_device.process_content('ENDED', 'Task OK', context, False))| 1 | define the parameters to pass to the API |
| 2 | create the new managed entity |
| 3 | store the ID of the new managed entity in the workflow instance context |
Microservice functions
Call a microservice CREATE/UPDATE/DELETE function
from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
from msa_sdk.order import Order
from msa_sdk.orchestration import Orchestration
from msa_sdk import util
import json
me_id = context['me'][3:] (1)
micro_service_vars_array = { (2)
"object_id": "12.1.1.1",
"mask": "255.255.255.0",
"gateway": "10.10.1.254"
}
object_id = "null"
route = {"routing": {object_id: micro_service_vars_array}} (3)
try:
ms_order = Order(me_id)
ms_order.command_execute('CREATE', route) (4)
except Exception as e:
ret = MSA_API.process_content('FAILED', f'CREATE ERROR: {str(e)}', context, True)
print(ret)| 1 | Read the ID of the managed entity from the context, assuming the variable 'me' type is Device |
| 2 | Build the Microservice JSON params for the CREATE operation of the microservice. |
| 3 | The value of the key should match the Microservice file name (stripped of the .xml file extension) |
| 4 | Call the CREATE for simple_firewall MS for each device (use UPDATE or DELETE for the other operations) |
The function command_execute is defined in order.py
Call a microservice CREATE on multiple managed entities
from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
from msa_sdk.order import Order
from msa_sdk import util
dev_var = Variables()
dev_var.add('object_id')
dev_var.add('service')
dev_var.add('src_ip')
dev_var.add('src_itf')
dev_var.add('dst_ip')
dev_var.add('dst_itf')
dev_var.add('firewalls.0.id') (1)
dev_var = Variables()
context = Variables.task_call(dev_var)
object_id = context['object_id']
micro_service_vars_array = {"object_id": context['object_id'],
"src_ip": context['src_ip'],
"src_mask": '255.255.255.255',
"dst_ip": context['dst_ip'],
"dst_mask": '255.255.255.255',
"src_itf": context['src_itf'],
"dst_itf": context['dst_itf'],
"action": 'deny',
"service": context['service']
}
simple_firewall = {"simple_firewall": {object_id: micro_service_vars_array}}
firewalls = context['firewalls']
for firewall in firewalls:
devicelongid = firewall['id'][-3:]
try:
order = Order(devicelongid)
order.command_execute('CREATE', simple_firewall)
except Exception as e:
ret = MSA_API.process_content('FAILED', f'CREATE ERROR: {str(e)}', context, True)
print(ret)
ret = MSA_API.process_content('ENDED',
f'IPTABLES RULE INITIALIZED',
context, True)
print(ret)| 1 | use a variable array typed as a Managed Entity |
Call a microservice IMPORT function
from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
from msa_sdk.order import Order
from msa_sdk.orchestration import Orchestration
from msa_sdk import util
import json
try:
order = Order(me_id) (1)
order.command_execute('IMPORT', {"routing":"0"}) (2)
order.command_objects_instances("routing") (3)
ms_instances = json.loads(order.content) (4)
except Exception as e:
ret = MSA_API.process_content('FAILED', f'IMPORT ERROR: {str(e)}', context, True)
print(ret)| 1 | initialize an Order object |
| 2 | execute the IMPORT of a microservice defined in a file routing.xml |
| 3 | get the microservice instances |
| 4 | store the instance in a variable to further reuse |
How to create you libraries of functions
When developing a workflow you will probably have de define some functions that will be used in multiple tasks.
In order to avoid code duplication and ease the maintenance of your workflow one option is to create a common folder at the same level as your other task folder and create a python file common.py where the shared functions will be defined.
my_workflow
|- my_workflow.xml
|- common
|- common.py
|- process_1
|- task1.py
common.py will contain the python code and can also import the Python SDK as well as other Python modules
from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
from msa_sdk import util
from msa_sdk.order import Order
from datetime import datetime
import time
import json
import typing
import copy
import requests
import ipaddress
import re
import pandas as Pandas
dev_var = Variables()
context = Variables.task_call(dev_var)
# Function: to convert
def printTable(myDict): (1)
df = Pandas.DataFrame.from_records(myDict)
return df.to_string()| 1 | a function that can be used in a task |
In the task implementation you need to add the following lines to import this common library
import os
import os.path
import sys
from pathlib import Path
from msa_sdk.variables import Variables
from msa_sdk.msa_api import MSA_API
currentdir = os.path.dirname(os.path.realpath(__file__)) (1)
parentdir = os.path.dirname(currentdir)
sys.path.append(parentdir)
from common.common import *
dev_var = Variables()
context = Variables.task_call(dev_var)
if data:
result = printTable(data) (2)
context['result'] = result
ret = MSA_API.process_content('ENDED', 'DONE', context, True)
print(ret)| 1 | include the path of common in the modules |
| 2 | call the method defined in the common library |
How to extend the SDK
Create a custom library of scripts
You can extend the SDK by adding your own scripts in the MSactivator™. The scripts have to be added in the container msa_dev, under the directory /opt/fmc_repository/Process/PythonReference/custom
In a workflow task, you can use the code below to import your custom scripts
import custom.myfileor
from custom.myfile import SOME_METHOD
You can create a git repository under /opt/fmc_repository/Process/PythonReference/custom with git init and set a remote to a remote repository to ease the management and versioning of your custom SDK library
|
Install additional Python modules
To install an additional Python package you need to log into the msa_dev container and execute
python3 -m pip install \
--install-option="--install-lib=/opt/fmc_repository/Process/PythonReference" PACKAGE (1)| 1 | PACKAGE is the name of the Python package to install |
To list the packages that are installed
python3 -m pip listMiscellaneous
Output messages to the process execution UI
When a task runs, it is often useful to be able to provide real time message update on the UI.
Output message from a task to the user interface
The code sample below shows how to do it.
from msa_sdk.orchestration import Orchestration
from msa_sdk.msa_api import MSA_API
import time
Orchestration = Orchestration(context['UBIQUBEID'])
async_update_list = (context['PROCESSINSTANCEID'],
context['TASKID'],
context['EXECNUMBER']) (1)
Orchestration.update_asynchronous_task_details(*async_update_list,
'going to sleep') (2)
time.sleep(2) (3)
Orchestration.update_asynchronous_task_details(*async_update_list,
'wake up') (4)| 1 | creates an array with the information about current process and task |
| 2 | update the UI with a message |
| 3 | execute some code |
| 4 | update the UI with another message |
Write debug message in the process log file
To write debugging messages in the process execution log file (msa-api container, under /opt/wildfly/logs ) you can use the function log_to_process_file
from msa_sdk import util
dev_var = Variables()
context = Variables.task_call(dev_var)
process_id = context['SERVICEINSTANCEID'] (1)
util.log_to_process_file(process_id, 'a debug message')| 1 | read the current process ID from the context |