FM / loaders

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

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

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
app.py	199	0644	edit dl rm
base.py	9147	0644	edit dl rm
default.py	1520	0644	edit dl rm
__init__.py	490	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/loaders/default.py (1520B)

"""The default loader used when no custom app has been initialized."""
import os
import warnings

from celery.exceptions import NotConfigured
from celery.utils.collections import DictAttribute
from celery.utils.serialization import strtobool

from .base import BaseLoader

__all__ = ('Loader', 'DEFAULT_CONFIG_MODULE')

DEFAULT_CONFIG_MODULE = 'celeryconfig'

#: Warns if configuration file is missing if :envvar:`C_WNOCONF` is set.
C_WNOCONF = strtobool(os.environ.get('C_WNOCONF', False))

class Loader(BaseLoader):
    """The loader used by the default app."""

def setup_settings(self, settingsdict):
        return DictAttribute(settingsdict)

def read_configuration(self, fail_silently=True):
        """Read configuration from :file:`celeryconfig.py`."""
        configname = os.environ.get('CELERY_CONFIG_MODULE',
                                    DEFAULT_CONFIG_MODULE)
        try:
            usercfg = self._import_config_module(configname)
        except ImportError:
            if not fail_silently:
                raise
            # billiard sets this if forked using execv
            if C_WNOCONF and not os.environ.get('FORKED_BY_MULTIPROCESSING'):
                warnings.warn(NotConfigured(
                    'No {module} module found! Please make sure it exists and '
                    'is available to Python.'.format(module=configname)))
            return self.setup_settings({})
        else:
            self.configured = True
            return self.setup_settings(usercfg)

"""The default loader used when no custom app has been initialized.""" import os import warnings from celery.exceptions import NotConfigured from celery.utils.collections import DictAttribute from celery.utils.serialization import strtobool from .base import BaseLoader __all__ = ('Loader', 'DEFAULT_CONFIG_MODULE') DEFAULT_CONFIG_MODULE = 'celeryconfig' #: Warns if configuration file is missing if :envvar:`C_WNOCONF` is set. C_WNOCONF = strtobool(os.environ.get('C_WNOCONF', False)) class Loader(BaseLoader): """The loader used by the default app.""" def setup_settings(self, settingsdict): return DictAttribute(settingsdict) def read_configuration(self, fail_silently=True): """Read configuration from :file:`celeryconfig.py`.""" configname = os.environ.get('CELERY_CONFIG_MODULE', DEFAULT_CONFIG_MODULE) try: usercfg = self._import_config_module(configname) except ImportError: if not fail_silently: raise # billiard sets this if forked using execv if C_WNOCONF and not os.environ.get('FORKED_BY_MULTIPROCESSING'): warnings.warn(NotConfigured( 'No {module} module found! Please make sure it exists and ' 'is available to Python.'.format(module=configname))) return self.setup_settings({}) else: self.configured = True return self.setup_settings(usercfg)