/usr/lib/python2.7/dist-packages/celery/app/annotations.py is in python-celery 3.1.20-1.
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 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | # -*- coding: utf-8 -*-
"""
celery.app.annotations
~~~~~~~~~~~~~~~~~~~~~~
Annotations is a nice term for moneky patching
task classes in the configuration.
This prepares and performs the annotations in the
:setting:`CELERY_ANNOTATIONS` setting.
"""
from __future__ import absolute_import
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']
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, 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]
def resolve_all(anno, task):
return (x for x in (_first_match(anno, task), _first_match_any(anno)) if x)
|