settei.presets.celery — Preset for Celery

class settei.presets.celery.WorkerConfiguration(config: Mapping[str, object] = {}, **kwargs)

The application object mixin which holds configuration for Celery.


Trigger the worker.on_loaded hooks. You should invoke this function when the Celery app is ready with the Celery app as argument. You may want to use celery.loaders.base.BaseLoader.on_worker_init

worker.on_loaded hook can be a Python code or list of module path.

When worker.on_loaded is a single string, it will be interpreted as Python code. The configuration and the Celery app is injected as self and app each:

on_loaded = """
print('Hello, world!')
print('self is configuration!: {}'.format(self))
print('app is celery app!: {}'.format(app))

When worker.on_loaded is a list of string, it will be interpreted as module paths:

on_loaded = [

The hook must receive two arguments, Configuration and celery.Celery:

def sample_hook(conf: Configuration, app: Celery):
    print('Hello, world!')
    print('conf is configuration!: {}'.format(conf))
    print('app is celery app!: {}'.format(app))
Parameters:app (celery.Celery) – a ready celery app

Changed in version 0.5.2: Hooks list added


The url of the broker used by Celery. See also Celery’s and Kombu’s docs about broker urls:


(typing.Mapping[str, object]) The configuration maping for worker that will go to Celery.conf.


The backend used by Celery to store task results. See also Celery’s docs about result backends:


(typing.Mapping[str, typing.Mapping[str, object]]) The schedule table for Celery Beat, scheduler for periodic tasks.

There’s some preprocessing before reading configuration. Since TOML doesn’t have custom types, you can’t represent timedelta or crontab values from the configuration file. To workaround the problem, it evaluates strings like 'f()' pattern if they are appeared in a schedule field.

For example, if the following configuration is present:

task = "tasks.add"
schedule = "timedelta(seconds=30)"  # string to be evaluated
args = [16, 16]

it becomes translated to:

    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': datetime.timedelta(seconds=30),  # evaluated!
        'args': (16, 16),

Note that although timedelta and crontab is already present in the context, you need to import things if other types. It can also parse and evaluate the patterns like 'module.path:func()'.

Also args fields are translated from array to tuple.

See also Celery’s docs about periodic tasks:

New in version 0.2.2.