Celery Bundle API

flask_unchained.bundles.celery

CeleryBundle

The Celery Bundle.

flask_unchained.bundles.celery.config

Config

Default configuration options for the Celery Bundle.

flask_unchained.bundles.celery.extensions

Celery

The Celery extension.

flask_unchained.bundles.celery.hooks

DiscoverTasksHook

Discovers celery tasks.

CeleryBundle

class flask_unchained.bundles.celery.CeleryBundle

The Celery Bundle.

name: str = 'celery_bundle'

The name of the Celery Bundle.

command_group_names = ['celery']

Click groups for the Celery Bundle.

Config

class flask_unchained.bundles.celery.config.Config

Default configuration options for the Celery Bundle.

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'

The broker URL to connect to.

CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'

The result backend URL to connect to.

CELERY_ACCEPT_CONTENT = ('json', 'pickle', 'dill')

Tuple of supported serialization strategies.

MAIL_SEND_FN(to=None, template=None, **kwargs)

If the celery bundle is listed after the mail bundle in unchained_config.BUNDLES, then this configures the mail bundle to send emails asynchronously.

The Celery Extension

class flask_unchained.bundles.celery.Celery(*args, **kwargs)

The Celery extension:

from flask_unchained.bundles.celery import celery

task(*args, **opts)

Decorator to create a task class out of any callable.

See Task options for a list of the arguments that can be passed to this decorator.

Examples:

@app.task
def refresh_feed(url):
    store_feed(feedparser.parse(url))

with setting extra options:

@app.task(exchange='feeds')
def refresh_feed(url):
    return store_feed(feedparser.parse(url))

Note:

App Binding: For custom apps the task decorator will return a proxy object, so that the act of creating the task is not performed until the task is used or the task registry is accessed.

If you’re depending on binding to be deferred, then you must not access any attributes on the returned object until the application is fully set up (finalized).

DiscoverTasksHook

class flask_unchained.bundles.celery.hooks.DiscoverTasksHook(unchained: flask_unchained.unchained.Unchained, bundle: Optional[flask_unchained.bundles.Bundle] = None)

Discovers celery tasks.

name: str = 'celery_tasks'

The name of this hook.

bundle_module_names: Union[List[str], Tuple[str, ...], None] = ['tasks']

The default module this hook loads from.

Override by setting the celery_tasks_module_names attribute on your bundle class.

process_objects(app: flask_unchained.flask_unchained.FlaskUnchained, objects: Dict[str, Any])

Implement to do stuff with discovered objects (eg, registering them with the app instance).

type_check(obj: Any)

Implement to determine which objects in a module should be processed by this hook.