Skip to main content

Tasks

All Swarm REST API resources use the following base URL:

https://swarm.unmand.app/


Trigger New Task

A new task can be created and queued at any time by invoking this endpoint. A task will always need an initial data packet and therefore it is expected to be included in the JSON payload of the request. If there is no initial data to provide, send an empty JSON object {}.

URL : /tasks/trigger

Method : POST

Query Parameters :

NameRequiredDescriptionDefault
environmentNoA task can be run in two environments: TEST or PROD.TEST

Example

from typing import Dict, Any
import os
import requests
from requests.auth import AuthBase

class TokenAuth(AuthBase):
"""Implements a custom authentication scheme."""

def __init__(self, token: str):
self.token: str = token

def __call__(self, r: requests.PreparedRequest):
"""Attach an API token to a custom auth header."""
r.headers['Authorization'] = self.token
return r

# Example request using TokenAuth class
API_TOKEN: str = os.environ['YOUR_API_TOKEN']
PAYLOAD: Dict[str, Any] = {
"first_name": "John",
"last_name": "Smith",
"amount": 100.02
}
PARAMS: Dict[str, str] = {
"environment": "PROD"
}
response: requests.Response = requests.post(
'https://swarm.unmand.app/tasks/trigger',
json=PAYLOAD,
params=PARAMS,
auth=TokenAuth(API_TOKEN)
)

Response

201 CREATED
{
"taskGuid": "T1209e5fd-eb1f-4eu4-a9f8-628326ed62a0"
}

List Tasks

Retrieve a list of tasks matching a specified query, limited to 50 results. The data in the fields will depend on what you have defined via the Unmand Portal.

URL : /projects/<PROJECT_GUID>/tasks

Method : GET

Query Parameters (optional) : &search=field1:5614523

NameRequiredDescriptionDefault
pageNoA cursor for pagination across multiple pages of results. Don’t include this parameter on the first call. The API returns 50 Task objects at a time by default.1
searchNoA search query string. The query string is checked for in the field data as well as the Task outcome and GUID
sort_orderNoOrder the results by Ascending or DescendingDescending
sort_fieldNoSort the results by the date the Task was Created or UpdatedUpdated
statusNoFilter the results by the Task status.All
environmentNoFilter the results by the Task environment: TEST or PRODAll

Example

import os
import requests
from requests.auth import AuthBase

class TokenAuth(AuthBase):
"""Implements a custom authentication scheme."""

def __init__(self, token: str):
self.token: str = token

def __call__(self, r: requests.PreparedRequest):
"""Attach an API token to a custom auth header."""
r.headers['Authorization'] = self.token
return r

# Example request using TokenAuth class
API_TOKEN: str = os.environ['YOUR_API_TOKEN']
PROJECT_GUID: str = "YOUR_PROJECT_GUID"
PARAMS: Dict[str, str] = {
"environment": "PROD",
"status": "SUCCESS"
}
response: requests.Response = requests.get(
f'https://swarm.unmand.app/projects/{PROJECT_GUID}/tasks',
params=PARAMS,
auth=TokenAuth(API_TOKEN)
)

Response

200 OK
{
"tasks": [
{
"guid": "Tb4ff2642-6655-48b4-8362-edc2c3f77255",
"status": "SUCCESS",
"outcome": "Task completed",
"created": "2022-02-17T00:23:08.899996+00:00",
"updated": "2022-02-17T00:25:06.090146+00:00",
"environment": "PROD",
"field1": "5614523",
"field2": "John Doe",
"field3": "01/01/2000",
"field5": "Melbourne",
"field6": "Male",
"field7": "High",
"field8": "$42.16",
"field9": "Yes",
},
{...},
{...}
],
"totalCount": 3
}

List Tasks (Version 2)

Retrieve a list of tasks matching a specified query, with advanced filtering options. This endpoint provides similar functionality to the original /projects/<PROJECT_GUID>/tasks endpoint but offers a different way of filtering the results. Limited to 50 results.

URL : /projects/<PROJECT_GUID>/tasks/v2

Method : `GET```

Query Parameters (optional) : ?searchSubject=field1&search__ilike=5614523

The following query parameters can be included in the request to customize the results:

ParameterDescriptionExample
pagePage number for pagination. Default value: 1.page=2
perPageNumber of results per page.perPage=10
environment__inFilter the results by task environment. Values can include TEST, PROD, or a combinationenvironment__in=TEST&environment__in=PROD
status__inFilter the results by task status.status__in=SUCCESS&status__in=QUEUED
searchSubjectRequired for search. Specifies the field to search in.searchSubject=field1
search__ilikeRequired for search. Search for a specific value using ILIKE (case-insensitive).search__ilike=5614523
dateSourceRequired for date range filter. Should be Created or Updated.dateSource=Updated
startDate__gteStart date for the date range filter.startDate__gte=2022-01-15T08:20:10.987654Z
endDate__lteEnd date for the date range filter.endDate__lte=2023-08-30T15:45:30.123456Z
sortFieldRequired for sorting. Field to sort the results by.sortField=Created
sort__asc or sort__descRequired for sorting. Order in which to sort the results.sort__desc=Updated

Note

  • For environment__in, status__in, and other multi-value filters, multiple values can be included by repeating the parameter (e.g., environment__in=TEST&environment__in=PROD).
  • For the search__ilike parameter, searchSubject is required to specify the field to search in. Search subjects include outcome for the task outcome, guid for the task identifier, and custom project task fields as field1, field2, etc.
  • dateSource should be one of Created or Updated.

Example

import os
import requests
from requests.auth import AuthBase

class TokenAuth(AuthBase):
"""Implements a custom authentication scheme."""

def __init__(self, token: str):
self.token: str = token

def __call__(self, r: requests.PreparedRequest):
"""Attach an API token to a custom auth header."""
r.headers['Authorization'] = self.token
return r

# Example request using TokenAuth class
API_TOKEN: str = os.environ['YOUR_API_TOKEN']
PROJECT_GUID: str = "YOUR_PROJECT_GUID"
PARAMS: Dict[str, str] = {
"environment__in": "PROD",
"status__in": ["SUCCESS", "FAILURE"]
}
response: requests.Response = requests.get(
f'https://swarm.unmand.app/projects/{PROJECT_GUID}/tasks/v2',
params=PARAMS,
auth=TokenAuth(API_TOKEN)
)

Response

200 OK
{
"tasks": [
{
"guid": "Tb4ff2642-6655-48b4-8362-edc2c3f77255",
"status": "SUCCESS",
"outcome": "Task completed",
"created": "2022-02-17T00:23:08.899996+00:00",
"updated": "2022-02-17T00:25:06.090146+00:00",
"environment": "PROD",
"field1": "5614523",
"field2": "John Doe",
"field3": "01/01/2000",
"field5": "Melbourne",
"field6": "Male",
"field7": "High",
"field8": "$42.16",
"field9": "Yes",
},
{...},
{...}
],
"totalCount": 3
}

Get Task

Retrieves a representation of the Task's current state. In addition to the Task's metadata, this notably also returns an array of Stage objects and the Task data packet.

URL : /tasks/<TASK_GUID>

Method : GET

Query Parameters : None

Example

from typing import Dict, Any
import os
import requests
from requests.auth import AuthBase

class TokenAuth(AuthBase):
"""Implements a custom authentication scheme."""

def __init__(self, token: str):
self.token: str = token

def __call__(self, r: requests.PreparedRequest):
"""Attach an API token to a custom auth header."""
r.headers['Authorization'] = self.token
return r

# Example request using TokenAuth class
API_TOKEN: str = os.environ['YOUR_API_TOKEN']
TASK_GUID: str = 'YOUR_TASK_GUID'
response: requests.Response = requests.get(
f'https://swarm.unmand.app/tasks/{TASK_GUID}',
auth=TokenAuth(API_TOKEN)
)

Response

200 OK
{
"task": {
"guid": "T7381f817-5cb5-4354-952c-66d80e12f9e6",
"status": "SUCCESS",
"created": "2022-02-23T04:03:07.652470+00:00",
"updated": "2022-02-23T04:04:42.279940+00:00",
"environment": "PROD",
"data": {
"amount": {
"value": 436.2,
"originalString": null
},
{...},
{...}
},
"stages": [
{
"guid": "0d8f06d6-69c0-4594-81b0-960d1a96b67e",
"name": "Update Quote",
"type": "SEQUENCE",
"status": "SUCCESS",
"started": "2022-02-23T04:03:17.478706",
"finished": "2022-02-23T04:03:47.824803",
"outcome_message": null
},
{...},
{...}
],
"outcome": "Policy bound",
"swarm_version": "dfd5e8a",
"rawData": {
"amount": 436.2,
...,
...,
}
}
}

The data property of the task object contains a pre-processed representation of the Task data that powers the Task Data editor in the Portal. The rawData property contains the actual Task data packet.


Resubmit Task

Resubmitting a task will run the job against the latest data in Task data and flow logic.

URL : /tasks/<TASK_GUID>/resubmit

Method : POST

Query Parameters : None

JSON Payload :

KeyRequiredTypeDescriptionDefault
forceNobooleanA resubmission is blocked if the stages in the flow have changed from when the task was last run. Set this to True to confirm the stage index provided is correct in the currently active flow versionFalse
stageYesnumberIndex of the stage you want to run from

Example

from typing import Dict, Any
import os
import requests
from requests.auth import AuthBase

class TokenAuth(AuthBase):
"""Implements a custom authentication scheme."""

def __init__(self, token: str):
self.token: str = token

def __call__(self, r: requests.PreparedRequest):
"""Attach an API token to a custom auth header."""
r.headers['Authorization'] = self.token
return r

# Example request using TokenAuth class
API_TOKEN: str = os.environ['YOUR_API_TOKEN']
TASK_GUID: str = 'YOUR_TASK_GUID'
PAYLOAD: Dict[str, Any] = {
"force": True,
"stage": 0,
}

response: requests.Response = requests.post(
f'https://swarm.unmand.app/tasks/{TASK_GUID}/resubmit',
json=PAYLOAD,
auth=TokenAuth(API_TOKEN)
)

Response

200 OK
{
"job_guid": "J7453g954-5cb5-4354-952c-54d80e13f9e6"
}