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/events.py (1326B)

"""Implementation for the app.events shortcuts."""
from contextlib import contextmanager

from kombu.utils.objects import cached_property

class Events:
    """Implements app.events."""

receiver_cls = 'celery.events.receiver:EventReceiver'
    dispatcher_cls = 'celery.events.dispatcher:EventDispatcher'
    state_cls = 'celery.events.state:State'

def __init__(self, app=None):
        self.app = app

@cached_property
    def Receiver(self):
        return self.app.subclass_with_self(
            self.receiver_cls, reverse='events.Receiver')

@cached_property
    def Dispatcher(self):
        return self.app.subclass_with_self(
            self.dispatcher_cls, reverse='events.Dispatcher')

@cached_property
    def State(self):
        return self.app.subclass_with_self(
            self.state_cls, reverse='events.State')

@contextmanager
    def default_dispatcher(self, hostname=None, enabled=True,
                           buffer_while_offline=False):
        with self.app.amqp.producer_pool.acquire(block=True) as prod:
            # pylint: disable=too-many-function-args
            # This is a property pylint...
            with self.Dispatcher(prod.connection, hostname, enabled,
                                 prod.channel, buffer_while_offline) as d:
                yield d

cmd: