Wir haben einige Blogposts über die Benutzung von PowerShell mit MOVEit Automation 2018 zur Automatisierung von Tasks geschrieben. Was aber, wenn Sie stattdessen Python benutzen möchten? Das ist auch kein Problem. Lesen Sie weiter…
In einem vorherigen Blogpost zur Verwendung der REST-API von MOVEit Automation 2018 haben wir PowerShell als Client verwendet. Diesmal sehen wir, wie man die Python-Skriptsprache benutzen kann, um die gleichen Tasks zu erstellen.
Wir werden den gleichen Task wie in dem PowerShell-Artikel als Beispiel verwenden, um einen angemessenen Vergleich zu haben - das Erstellen und Überwachen von Tasks. Der Task beginnt mit Dateien in einem Ordner auf meinem MOVEit 2018-Server. Ich möchte alle diese Dateien in einen Amazon S3-Bucket hochladen. Wir werden sehen, wie dieser Task mit Python erstellt und gestartet werden kann.
Authentifizierung
Mit Bezug auf einen vorherigen Artikel zur Authentifizierung zur MOVEit Automation-REST-API verwende ich denselben Ansatz, aber diesmal mit Python anstelle von PowerShell.
Zuerst definiere ich alle Variablen, die die verschiedenen URLs, Pfade und Authentifizierungsinformationen repräsentieren, die ich später brauche.
base_uri = 'https://<YourMoveITAutomationServerNameHere>'
token_path = '/webadmin/api/v1/token'
uri_path = '/webadmin/api/v1/tasks'
api_endpoint = base_uri + token_path
tasks_url = base_uri + uri_path
userName = '<myusername>'
password = '<mypassword>'
Als nächstes muss ich einen Authentifizierungs-Token erhalten. In Python kann man dafür ein Modul namens requests verwenden, das eine HTTP GET-Anfrage an die API senden wird. Das müssen wir importieren. Da ich mit einem MOVEit -Server arbeite, der über ein selbstsigniertes Zertifikat verfügt, muss ich Python mitteilen, dass die Sicherheitswarnung ignorieren werden soll, die ich mit der Methode disable_warnings() in der urllib3 library erhalte.
Zum Schluss erstelle ich ein Verzeichnis, das alle Authentifizierungsattribute enthält, die ich an den Server übergeben muss, wenn die Anfrage ausgeführt wird.
import requests
# Is this a self signed cert enable this to ignore SSL errors
requests.packages.urllib3.disable_warnings()
payload = {'username': userName, 'password': password, 'grant_type': 'password'}
Jetzt kann man die Anfrage senden und den Token, der abgerufen wurde, der Variablen my_token zuweisen.
r = requests.post(api_endpoint, data=payload, verify=False)
#Auth token to be used by successive requests
my_token = r.json()['access_token']
Erstellen des Tasks
Wenn ich mir die Swagger-Benutzeroberfläche in MOVEit ansehe, sehe ich eine Operation, die genau richtig für meine Zwecke ist. Die Beschreibung "adds a new task" ist selbsterklärend.
Wenn ich hineinklicke, sehe ich, dass der API JSON gegeben werden muss, das den Task, welcher erstellt wird, repräsentieren wird. Mit Hilfe des zur Verfügung gestellten JSON Beispiels habe ich JSON-Code geschrieben, welcher den Task repräsentiert, den ich ausführen möchte.
{
"Info": {
"Description": "This task moves some files locally to an AWS S3 bucket on demand."
},
"steps": [
{
"Source": {
"HostID": "0",
"Path": "C:\\Temp",
"Type": "FileSystem"
}
},
{
"Destination": {
"HostID": "730111199",
"Type": "S3"
}
}
],
"Name": "Copy Files to S3 Bucket",
"Active": 1
}
Das so erstellte JSON kopiere ich in eine Datei namens DemoTask.json auf meinem lokalen Computer. Wenn ich das gemacht habe, muss ich diesen JSON in eine Zeichenkette bringen, um diese schließlich an die API zu geben.
Mit der load ()-Methode für das json-Modul kann ich ein JSON-Objekt erstellen, wobei ich zuerst daran denken muss, mein json-Modul zu importieren.
import json
#convert json file to python dictionary
with open('/Users/adam/Desktop/DemoTask.json') as json_data:
job_json = json.load(json_data)
Nun habe ich alles, was ich brauche, um einen API-Aufruf an die MOVEit-API zu machen, um diese Anfrage zu erstellen.
#Create new Task
new_task = requests.post(tasks_url, headers={'Content-Type': 'application/json;charset=UTF-8', 'Authorization': 'Bearer {}'.format(my_token)},json=job_json, verify=False)
Ich sollte nun einen Task namens Copy Files to S3 Bucket haben. Um sicher zu gehen, stelle ich eine Anfrage an die API mit dem Namen des Tasks.
task_name = 'Copy Files to S3 Bucket'
query_url = tasks_url + '?name=' + task_name
task_query = requests.get(query_url, headers={'Content-Type': 'application/json;charset=UTF-8', 'Authorization': 'Bearer {}'.format(my_token)}, verify=False)
task_json = json.loads(task_query.content)
Ich schaue mir den zurückerhaltenen JSON an, um zu überprüfen, ob dies der richtige Task ist.
>>> task_json
{u'Info': {u'Notes': u'', u'Description': u'This task moves some files locally to an AWS S3 bucket on demand.'}, u'Group': [], u'Name': u'Copy Files to S3 Bucket', u'UseDefStateCaching': 1, u'CacheNames': u'random', u'TT': u'', u'NextActions': {}, u'NextEID': 13, u'AR': 0, u'steps': [{u'Source': {u'UseDefRetryTimeoutSecs': 1, u'ExFo': u'', u'NewFilesOnly': 0, u'DeleteOrig': 0, u'RetryCount': 0, u'SearchSubdirs': 0, u'Type': u'FileSystem', u'Unzip': 0, u'ExFile': u'', u'Path': u'C:\\Temp', u'RetryTimeoutSecs': 0, u'HostID': u'0', u'DelRename': 1, u'FileMask': u'*.*', u'UseDefRescanSecs': 1, u'RenameTo': u'', u'ID': u'11', u'MxBy': 0, u'RescanSecs': 0, u'RetryIfNoFiles': 0, u'UseDefRetryCount': 1, u'MxFi': 0}}, {u'Destination': {u'UseDefRetryTimeoutSecs': 1, u'Zip': 0, u'OverwriteOrig': 1, u'RetryCount': 0, u'Type': u'S3', u'Username': u'', u'UseRelativeSubdirs': 1, u'UseDefBucket': 1, u'Path': u'', u'Password': u'', u'RetryTimeoutSecs': 0, u'HostID': u'730111199', u'ConnTimeoutSecs': 0, u'Bucket': u'', u'UseDefUser': 1, u'UseDefRescanSecs': 1, u'UseOrigName': 1, u'ID': u'12', u'ForceDir': 1, u'FileName': u'[OrigName]', u'RescanSecs': 0, u'UseDefConnTimeoutSecs': 1, u'UseDefRetryCount': 1}}], u'Schedules': {u'Schedule': []}, u'Active': 1, u'ID': u'218513542'}
Erstellung des Tasks
Man kann nun die Aufgabe starten, indem man die entsprechende Start-URL erstellt und diese verwendet, um den API-Aufruf erneut zu senden. Im Folgenden überprüfe ich das content-Attribut der Antwort, um zu sehen, ob der Task gestartet wurde.
task_id = task_json['items'][0]['ID']
#Create start task url
start_url = "{}/{}/start".format(tasks_url, task_id)
#Initiate job
start_result = requests.post(start_url, headers={'Content-Type': 'application/json;charset=UTF-8', 'Authorization': 'Bearer {}'.format(my_token)}, verify=False)
>>> print(start_result.content)
{"nominalStart":"2018-04-25 20:07:53.54"}
Tipp: Laden Sie sich kostenfrei eine Testversion von MOVEit Automation, Software für die Automatisierung der Dateiübertragung, herunter.