celery.py 文件源码

def worker_config(self) -> typing.Mapping[str, object]:
        """(:class:`typing.Mapping`\ [:class:`str`, :class:`object`])
        The configuration maping for worker that will go to :attr:`Celery.conf
        <celery.Celery.conf>`.

        """
        raw_config = self.config.get('worker', {})
        if isinstance(raw_config, collections.abc.Mapping):
            celery_config = {k.upper(): v for k, v in raw_config.items()}
        else:
            celery_config = {}
        celery_config.update(
            BROKER_URL=self.worker_broker_url,
            CELERY_RESULT_BACKEND=self.worker_result_backend,
            CELERYBEAT_SCHEDULE=self.worker_schedule
        )
        return celery_config