celery-dispatcher 1.1.3
An extension for celery to dispatch large amount of subtasks within a main task
Requires Python: >=3.6

# celery-dispatcher

An extension for celery to dispatch large amount of subtasks within a main task, and process the results separately.

## Installation

```
pip install celery-dispatcher
```

## Usage

> NOTICE: `celery-dispatcher` use tls to store running info, so the pool implementation using coroutines(like eventlet/gevent) can not be used

Firstly, yield subtask and its parameters in tuple from the main task by following order:

1. `task`: signature
2. `args`: tupple/list
3. `kwargs`: dict
4. `options`: dict

Then register the result handler for each subtask using signal:

```
from celery import shared_task
from celery_dispatcher import dispatch
from celery_dispatcher.signals import subtask_success

@shared_task
def sqrt(i):
    return i * i

@dispatch
@shared_task
def calc():
    for i in range(10):
        yield sqrt, (i,)

@subtask_success.connect(sender=calc)
def handle_result(root_id, task_id, retval, **kwargs):
    print(retval)
```

Or register in the decorator directly:

```
from celery import shared_task
from celery_dispatcher import dispatch

@shared_task
def sqrt(i):
    return i * i

def handle_result(root_id, task_id, retval, **kwargs):
    print(retval)

@dispatch(receiver=handle_result)
@shared_task
def calc():
    for i in range(10):
        yield sqrt, (i,)
```

If you want to revoke all subtasks, call revoke with specific signal:

```
import signal
from celery import shared_task
from celery_dispatcher import dispatch

@shared_task
def subtask(i):
    ...

@dispatch
@shared_task
def maintask():
    for i in range(10):
        yield subtask, (i,)

result = maintask.delay()
result.revoke(terminate=True, signal=signal.SIGUSR1)
```

## Options

The `dispatch` accepts the following parameters:

- `options`: dict
- `receiver`: callable
- `backend`: str
- `auto_ignore`: bool

## General settings

### dispatcher_result_backend

Default: No result backend enabled by default.

The backend used to store subtask info. Can be one of the following:

- redis: Use [Redis](https://redis.io/) to store the results. See [Redis backend settings](https://docs.celeryproject.org/en/stable/userguide/configuration.html#conf-redis-result-backend).

### dispatcher_batch_size

Default: 100

The batch size of subtask dispatching, or the result retrieving.

### dispatcher_poll_size

Default: 100

The queue size of subtasks in result retrieving. `celery-dispatcher` put the unfinished subtasks into a queue, polling it continuously for completion of subtasks.

### dispatcher_poll_timeout

Default: 10

The default timeout for polling subtasks.

### dispatcher_subtask_timeout

Default: 3600

The default timeout in seconds before `celery-dispatcher` gives up retrieving the result of each subtask.

### dispatcher_failure_on_subtask_timeout

Default: False

Whether raise exception when subtask timeout.

### dispatcher_failure_on_subtask_exception

Default: False

Whether raise exception from subtask.

### dispatcher_progress_update_frequency

Default: 1

The number of steps to record. `celery-dispatcher` only actually store the updated progress in the background at most every N steps.