/usr/share/pyshared/celery/app/annotations.py is in python-celery 2.5.3-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | from __future__ import absolute_import
from ..utils import firstmethod, instantiate, mpromise
_first_match = firstmethod("annotate")
_first_match_any = firstmethod("annotate_any")
class MapAnnotation(dict):
def annotate_any(self):
try:
return dict(self["*"])
except KeyError:
pass
def annotate(self, task):
try:
return dict(self[task.name])
except KeyError:
pass
def prepare(annotations):
"""Expands the :setting:`CELERY_ANNOTATIONS` setting."""
def expand_annotation(annotation):
if isinstance(annotation, dict):
return MapAnnotation(annotation)
elif isinstance(annotation, basestring):
return mpromise(instantiate, annotation)
return annotation
if annotations is None:
return ()
elif not isinstance(annotations, (list, tuple)):
annotations = (annotations, )
return map(expand_annotation, annotations)
|