Skip to content

Tasks

Lime Task provides the functionality to run asynchronous tasks, declare/create and start a task as well as asking for its status.

In order to gain the best performance it's a good practice to base your solutions on tasks. Any action that takes longer than a normal REST request should be realized as a task. Same good use cases are e.g. bulk updates, heavy processing, reporting or integrations.

In comparison to a REST request the execution is handled asynchronously. That means, that whenever a task is started, it's put in a queue and waits for an available worker to be executed. For that reason Lime Task can't promise any execution times just that the task is eventually taken care of. During that time a task owner can ask for the current status, which might be PENDING, STARTED, SUCCESS or FAILURE. Whenever the execution of a task is completed, the status object can also contain a result.

Generate the tasks module

To add the tasks module to your package, change directory to your package directory and run this command:

lime-project generate tasks

If you're working on Linux, make sure your package is installed on the taskhandler container.

Declare a task

The task decorator provides you with a lime application based on the application name you use in the send_task method. So you have to be aware of the fact, that the first argument of your task function is ALWAYS a lime application

from lime_application import LimeApplication
from lime_task import task


@task
def get_lime_company(
    application: LimeApplication, company_id: int, properties: dict
) -> dict:
    company = _get_company_object(application, company_id)
    description = _concat_description(company, properties)

    return {
        "description": description,
    }

As mentioned before a task status can contain a result after the execution is done. This is whatever this task function returns. The result should be rather considered as a summary of the task than the affected data.

Example of a good result

{
    "result": "updated the sales representative of 50 deal entities to coworker with id 1234",
    "id": "generated-id",
    "status": "SUCCESS"
}

Example of a bad result

{
    "result": {
        "deals": [
            {
                "id": 432,
                "salesRepresentative": 1234
            },
            {
                "id": 321,
                "salesRepresentative": 1234
            }
        ]
    },
    "id": "generated-id",
    "status": "SUCCESS"
}

Send a task

Use the send_task method from the lime_task package in order to start a task. You can either do that in a separate module in your package like this:

from lime_task import send_task

task_name = "my_package.tasks.lime.get_lime_company"  # task function name including the full module path
task_status = send_task(task_name, my_limeapp, 1001)

print("status:", task_status.status)
print("result:", task_status.result)

...or adjust the provided POST request in the generated task_endpoint.py. If you're working on Linux and want to use the endpoint, remember to also install the package on the appserver container.

Get status

You can get the current status of a task if you know the task id.

Note

Make sure the code is bootstrapped, otherwise the service locator and task status service will not be available.

Running this in the context of an endpoint, event handler or a task handler works since they are bootstrapped. The task status service is accessible as a function argument or class attribute of the type TaskStatusService.

Deprecated Method

The older lime_task.get_status(task_id, app_name) function is deprecated. Use TaskStatusService instead for new code.

Version Requirement

TaskStatusService was introduced in lime-core version 25.315.0 (September 2025), available in lime-crm version 2.1135.0 and later. For earlier versions, you must use the deprecated lime_task.get_status() function.

import lime_task
import lime_webserver.webserver


class TaskStatusResource(lime_webserver.webserver.LimeResource):
    task_status_service: lime_task.TaskStatusService

    def get(self, task_id: str):
        try:
            task_status = self.task_status_service.get(task_id)
        except lime_task.NotFoundError:
            return f"Task with id {task_id!r} was not found", 404
        except lime_task.TaskExecutionError as e:
            return {
                "id": task_id,
                "status": "FAILURE",
                "result": f"Task execution error: {str(e)}"
            }, 500

        return {
            "status": task_status.to_dict(),
        }

Task Status Response Format

The task_status.to_dict() method returns a standardized response:

{
  "id": "task-uuid-here",
  "status": "SUCCESS",
  "result": "Task completed: processed 42 companies"
}

Available Status Values

Tasks can have the following status values: - PENDING - Task is waiting to be processed - STARTED - Task is currently being executed
- SUCCESS - Task completed successfully - FAILURE - Task failed during execution - REVOKED - Task was cancelled - RETRY - Task is being retried after failure

Task events

After a task has run an event is dispatched on the message queue.

The following routing keys are used:

  • core.task.{task_name}.success.v1 if the task successfully ran
  • core.task.{task_name}.failure.v1 if the task failed

Hint

Build an event handler that listens to errors in your scheduled tasks. That will help you get notified and react properly e.g. by sending out an email when the task couldn't be completed.

Acknowledge a task

When working with tasks in Lime CRM, you have the option to choose between two types of acknowledgments: early acknowledgment (early ack) and late acknowledgment (late ack). These acknowledgments affect how the task execution is handled and how the task's state is managed. Early acknowledgment acknowledges the task as soon as it is received, while late acknowledgment waits for the worker to confirm task completion before acknowledging it.

The default behavior for a task in in Lime CRM is to do early acknowledgement. You can change that by writing acks_late=True in the task decorator:

import logging

from lime_task import task

logger = logging.getLogger(__name__)


@task(acks_late=True)
def my_task(app, *args, **kwargs):
    logger.info("Executing my_task with args: %s, kwargs: %s", args, kwargs)

When using late acknowledgment in task processing, it is important to ensure that the task is idempotent. Idempotence means that executing the task multiple times will have the same outcome as executing it once. Please beware, that writing an idempotent task can be challenging, especially when dealing with external systems and potential failures. It's essential to handle various scenarios appropriately to ensure the task behaves correctly and doesn't negatively impact the overall performance of your solution. When an external system is down or inaccessible, it's important to implement appropriate error handling mechanisms, otherwise there is a high risk that your task will be re-run forever and block other tasks in the system from running.

We prefer early ack

We recommend using early acknowledgment for easier task creation, as late acknowledgment tasks can run indefinitely and potentially block other tasks in the system.