This document describes the current stable version of Celery (4.0). For development docs, go here.

Source code for celery.app.annotations

# -*- coding: utf-8 -*-
"""Task Annotations.

Annotations is a nice term for monkey-patching task classes
in the configuration.

This prepares and performs the annotations in the
:setting:`task_annotations` setting.
"""
from __future__ import absolute_import, unicode_literals
from celery.five import string_t
from celery.utils.functional import firstmethod, mlazy
from celery.utils.imports import instantiate

_first_match = firstmethod('annotate')
_first_match_any = firstmethod('annotate_any')

__all__ = ['MapAnnotation', 'prepare', 'resolve_all']


[docs]class MapAnnotation(dict): """Annotation map: task_name => attributes."""
[docs] def annotate_any(self): try: return dict(self['*']) except KeyError: pass
[docs] def annotate(self, task): try: return dict(self[task.name]) except KeyError: pass
[docs]def prepare(annotations): """Expand the :setting:`task_annotations` setting.""" def expand_annotation(annotation): if isinstance(annotation, dict): return MapAnnotation(annotation) elif isinstance(annotation, string_t): return mlazy(instantiate, annotation) return annotation if annotations is None: return () elif not isinstance(annotations, (list, tuple)): annotations = (annotations,) return [expand_annotation(anno) for anno in annotations]
[docs]def resolve_all(anno, task): """Resolve all pending annotations.""" return (x for x in (_first_match(anno, task), _first_match_any(anno)) if x)