Skip to main content

Getting and using Tasks

Getting Tasks​

This function retrieves all the pending Tasks in a stage satisfying the search filters and completes them after processing.

from abstra.tasks import get_tasks

pending_tasks = get_tasks()
for task in pending_tasks:
    # do some processing
    task.complete()

This function returns a list of instances of the class Task.

Querying Tasks​

You can search pending tasks by filtering them with the where parameter. The where parameter is a dictionary that contains the search filters.

from abstra.tasks import get_tasks

pending_tasks = get_tasks(
    limit=100, 
    offset=20, 
    where={
        "name": "Michael Scott",
        "company": "Dunder Mifflin"
    }
)
info

The get_tasks function has a maximum limit of 10,000 records per request. If you need to retrieve more than 10,000 Tasks, use the iter_tasks function instead, which handles pagination automatically for large datasets.

Using Task data​

Using keys​

You can access directly the value of a key in the payload by using task[key].

print(task["name"]) # Michael Scott
print(task["company"]) # Dunder Mifflin

Getting complete payload​

You can access the full task payload by using task.payload.

task.payload
'''
Returns:
{
    "name": "Michael Scott",
    "company": "Dunder Mifflin",
}
'''

Task built-in data​

You can also retrieve the task id and type with task.id and task.type:

print(task.id) # 23786870-5603-4765-af7c-bcd959cdff2f
print(task.type) # approved

Locking a Task​

You can lock a task to prevent other processes from completing it while you are processing it. This is useful when you want to ensure that only one process is working on a task at a time.

To lock a task, you may use the with statement, which will automatically lock the task when entering the block and unlock it when exiting the block.

from abstra.tasks import get_trigger_task

task = get_trigger_task()
with task:
    # do anything

Alternatively, you can use the task.lock() method to lock the task and task.complete() to unlock it.

from abstra.tasks import get_tasks

pending_tasks = get_tasks()
for task in pending_tasks
    # lock the task
    task.lock()
    # do some processing
    # unlock the task
    task.complete()