#1157: Writing custom automation plugins with Python
Отредактирована: 102 дня назадYOU ARE USING THE RUNTIME SCRIPT MECHANISM THAT RUNS FROM ROOT WITH NO SAFEGUARDS AT YOUR OWN DISCRETION.
LICENSOR EXPRESSLY DISCLAIMS ALL WARRANTIES, WHETHER EXPRESS, IMPLIED, OR STATUTORY, INCLUDING BUT NOT LIMITED TO THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT.
TO THE FULLEST EXTENT PERMITTED BY LAW, IN NO EVENT SHALL LICENSOR BE LIABLE FOR ANY INDIRECT,
INCIDENTAL, SPECIAL, CONSEQUENTIAL, OR PUNITIVE DAMAGES, OR ANY LOSS OF PROFITS OR REVENUE, ARISING FROM
OR RELATED TO THIS AGREEMENT OR YOUR USE OF THE MECHANISM.
Intro
Runtime script engine allows you to write and use arbitrary Python code to run automations triggered by an event,
by a schedule, or even manually by a user via a web-form.
Runtime scripts can be uploaded and set up by Administrator in Settings > Automation and routing > Scripts
Security and performance considerations
Scripts run inside a container, so there is no access for them to the hardware node / virtual machine where Swarmica
is installed, however, the scripts have access to the shared volumes, and the main database used by Swarmica, so
be cautions about running potentially dangerous operations, such as deleting and modifying any data or files.
Also, the memory/processor resources are shared with the whole Swarmica instance, that's why it's strongly recommended
to minimize importing libraries and modules to only the necessary minimal set of functions.
E.g. instead of importing the whole module import datetime import only the necessary function you need:
from datetime import timedelta.
Runtime engine overview
There are few ways a script can be launched:
[Event Trigger] ---> +context ---\
\
[Schedule] --------> +context------\
>----> Runner.run(kwargs={data: **params, **context})
[CLI] -------------> +params-------/
/
[API/WEB] ---------> +params-----/
So, basically, if the runtime script triggered by an Event Trigger or Schedule, the event data
and custom context data configured in corresponding trigger / schedule is passed to the script
parameters.
Say, the script is triggered by UserEvent, so the context can be parsed as follows:
class Runner(BaseRunner):
def run(self, *args, **kwargs):
data = kwargs.get('data', {})
if not data or 'event_id' not in data:
logger.error(f"No event_id passed, exiting")
return
event = UserEvent.objects.filter(id=data['event_id']).first()
if not event:
logger.error(f"No event with {data['event_id']} found, exiting")
return
If we configure the Event Trigger to pass additional context to the script, e.g. list of emails
to receive an alert when a user account is changed, then it too will be propagated to the script.
Given that Event Trigger custom context looks like this:
{
"recipients": [
"user1@domain.tld",
"user2@domain.tld"
]
}
You can now access the variable value inside the script as follows:
data = kwargs.get('data', None)
if data:
recipients = data.get('recipients', None)
You can also configure the script to have it's own web form, that can accept, validate
and pass parameters to the script.
Say, we write a script that generates a custom Excel report and sends it to the list of email
recepients. And we want users to pass report start and end dates, and the comma-separated list
of recepients.
In Swarmica's web interface in the script's settings we configure the following webform params:
[
{
"name": "start_date",
"type": "datetime",
"readonly": false,
"displayName": "Report start date"
},
{
"name": "end_date",
"type": "datetime",
"readonly": false,
"displayName": "Report end date"
},
{
"name": "recipients",
"type": "text",
"displayName": "Comma-separated recipients"
}
]
So now we can access these parameters in the script as follows:
params = kwargs.get('data', {})
if params:
if 'recipients' in params:
recipients = params.get('recipients')
if recipients:
recipients = [x.strip() for x in recipients.split(',') if '@' in x]
if not recipients:
logger.info("Input error: empty/invalid list of 'recipients' specified")
return
start_date = params.get('start_date', None)
if isinstance(start_date, str) and start_date.strip():
report_start_date = parse_datetime(start_date)
end_date = params.get('end_date', None)
if isinstance(end_date, str) and end_date.strip():
report_end_date = parse_datetime(end_date)
if not report_end_date:
report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
if not report_start_date:
report_start_date = report_end_date - timedelta(days=1)
Accessing Swarmica's Objects
Typically, you'd need to manipulate models and querysets for Swarmica's objects within your scripts.
Since Swarmica uses Django as the framework, most of model and queryset methods work here as well.
See the complete reference of the Django model and queryset methods on official Doc Site:
Most of objects you may probably need are located in core package, except for User objects - those
are in swarmica_auth package.
Also, you would typically want to import general constants Swarmica use as follows:
from swarmica import defs
print(defs.USER_ROLES_INTERNAL)
Let's see some typical operations in action.
Working with a set of Tickets
Let's print out ticket ID, subject and assignee's name for all new and open tickets, created between
provided start and end date:
from swarmica import defs
from core.models import Ticket
class Runner(BaseRunner):
def run(self, *args, **kwargs):
params = kwargs.get('data', {})
if not params:
return
start_date = params.get('start_date', None)
if isinstance(start_date, str) and start_date.strip():
report_start_date = parse_datetime(start_date)
end_date = params.get('end_date', None)
if isinstance(end_date, str) and end_date.strip():
report_end_date = parse_datetime(end_date)
if not report_end_date:
report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
if not report_start_date:
report_start_date = report_end_date - timedelta(days=1)
tickets = Ticket.objects.filter(created_at__range=(report_start_date,report_end_date),
status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN])
for ticket in ticket:
print(f"#{ticket.id}: {ticket.subject} ({ticket.assignee.name})")
Working with Users
Let's turn all email notifications off for blocked users:
from swarmica import defs
from swarmica_auth.models import User
class Runner(BaseRunner):
def run(self, *args, **kwargs):
users = User.objects.filter(role=defs.USER_ROLES_BLOCKED)
for user in users:
user.email_notifications.clear()
Available 3rd-party imports:
There is a number of pre-installed libraries, available for you to import and use in your scripts.
See the list of available methods in the corresponding library's documentation:
import re # To work with regexp
import os # To work with OS things like file path, etc
import bs4 # To parse HTML, XML, etc. with BeautifulSoup
import csv # To read and write CSV
import json # To parse JSON
import time # To work with current time, e.g. generate timestamps
import pandas # To work with datasets
import urllib # To parse and write URL parameters, urlencode, etc.
import dateutil # To work with dates, e.g. parse date from string
import datetime # To work with datetime objects, e.g. calculate durations, etc.
import openpyxl # To read and write Excel
import requests # To work with remote APIs and webhooks