FM / app

/opt/imh-python/lib/python3.9/site-packages/celery/app/

/opt/imh-python/lib/python3.9/site-packages/celery/app

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
amqp.py	23236	0644	edit dl rm
annotations.py	1445	0644	edit dl rm
autoretry.py	2506	0644	edit dl rm
backends.py	2746	0644	edit dl rm
base.py	50584	0644	edit dl rm
builtins.py	6673	0644	edit dl rm
control.py	29171	0644	edit dl rm
defaults.py	15266	0644	edit dl rm
events.py	1326	0644	edit dl rm
log.py	9067	0644	edit dl rm
registry.py	2001	0644	edit dl rm
routes.py	4527	0644	edit dl rm
task.py	43794	0644	edit dl rm
trace.py	27551	0644	edit dl rm
utils.py	13160	0644	edit dl rm
__init__.py	2430	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/app/annotations.py (1445B)

"""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 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):
    """Annotation map: task_name => attributes."""

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):
    """Expand the :setting:`task_annotations` setting."""
    def expand_annotation(annotation):
        if isinstance(annotation, dict):
            return MapAnnotation(annotation)
        elif isinstance(annotation, str):
            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):
    """Resolve all pending annotations."""
    return (x for x in (_first_match(anno, task), _first_match_any(anno)) if x)

"""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 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): """Annotation map: task_name => attributes.""" 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): """Expand the :setting:`task_annotations` setting.""" def expand_annotation(annotation): if isinstance(annotation, dict): return MapAnnotation(annotation) elif isinstance(annotation, str): 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): """Resolve all pending annotations.""" return (x for x in (_first_match(anno, task), _first_match_any(anno)) if x)