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/registry.py (2001B)

"""Registry of available tasks."""
import inspect
from importlib import import_module

from celery._state import get_current_app
from celery.app.autoretry import add_autoretry_behaviour
from celery.exceptions import InvalidTaskError, NotRegistered

__all__ = ('TaskRegistry',)

class TaskRegistry(dict):
    """Map of registered tasks."""

NotRegistered = NotRegistered

def __missing__(self, key):
        raise self.NotRegistered(key)

def register(self, task):
        """Register a task in the task registry.

The task will be automatically instantiated if not already an
        instance. Name must be configured prior to registration.
        """
        if task.name is None:
            raise InvalidTaskError(
                'Task class {!r} must specify .name attribute'.format(
                    type(task).__name__))
        task = inspect.isclass(task) and task() or task
        add_autoretry_behaviour(task)
        self[task.name] = task

def unregister(self, name):
        """Unregister task by name.

Arguments:
            name (str): name of the task to unregister, or a
                :class:`celery.app.task.Task` with a valid `name` attribute.

Raises:
            celery.exceptions.NotRegistered: if the task is not registered.
        """
        try:
            self.pop(getattr(name, 'name', name))
        except KeyError:
            raise self.NotRegistered(name)

# -- these methods are irrelevant now and will be removed in 4.0
    def regular(self):
        return self.filter_types('regular')

def periodic(self):
        return self.filter_types('periodic')

def filter_types(self, type):
        return {name: task for name, task in self.items()
                if getattr(task, 'type', 'regular') == type}

def _unpickle_task(name):
    return get_current_app().tasks[name]

def _unpickle_task_v2(name, module=None):
    if module:
        import_module(module)
    return get_current_app().tasks[name]

"""Registry of available tasks.""" import inspect from importlib import import_module from celery._state import get_current_app from celery.app.autoretry import add_autoretry_behaviour from celery.exceptions import InvalidTaskError, NotRegistered __all__ = ('TaskRegistry',) class TaskRegistry(dict): """Map of registered tasks.""" NotRegistered = NotRegistered def __missing__(self, key): raise self.NotRegistered(key) def register(self, task): """Register a task in the task registry. The task will be automatically instantiated if not already an instance. Name must be configured prior to registration. """ if task.name is None: raise InvalidTaskError( 'Task class {!r} must specify .name attribute'.format( type(task).__name__)) task = inspect.isclass(task) and task() or task add_autoretry_behaviour(task) self[task.name] = task def unregister(self, name): """Unregister task by name. Arguments: name (str): name of the task to unregister, or a :class:`celery.app.task.Task` with a valid `name` attribute. Raises: celery.exceptions.NotRegistered: if the task is not registered. """ try: self.pop(getattr(name, 'name', name)) except KeyError: raise self.NotRegistered(name) # -- these methods are irrelevant now and will be removed in 4.0 def regular(self): return self.filter_types('regular') def periodic(self): return self.filter_types('periodic') def filter_types(self, type): return {name: task for name, task in self.items() if getattr(task, 'type', 'regular') == type} def _unpickle_task(name): return get_current_app().tasks[name] def _unpickle_task_v2(name, module=None): if module: import_module(module) return get_current_app().tasks[name]