Skip to content

climate_ref_celery.app #

Celery app creation

create_celery_app(name) #

Create a Celery app

This function creates a new Celery app with the given name and configuration module. The configuration module is loaded from the environment variable CELERY_CONFIG_MODULE which defaults to climate_ref_celery.celeryconf.dev if not set.

Source code in packages/climate-ref-celery/src/climate_ref_celery/app.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def create_celery_app(name: str) -> Celery:
    """
    Create a Celery app

    This function creates a new Celery app with the given name and configuration module.
    The configuration module is loaded from the environment variable `CELERY_CONFIG_MODULE`
    which defaults to `climate_ref_celery.celeryconf.dev` if not set.
    """
    register_serialisation()

    app = Celery(name)
    app.config_from_envvar("CELERY_CONFIG_MODULE")

    return app

setup_logging_handler(loglevel, **kwargs) #

Set up logging for the Celery worker using the celery signal

Source code in packages/climate-ref-celery/src/climate_ref_celery/app.py 
36
37
38
39
40
41
42
43
@setup_logging.connect
def setup_logging_handler(loglevel: int, **kwargs: Any) -> None:  # pragma: no cover
    """Set up logging for the Celery worker using the celery signal"""
    # We ignore the format passed by Celery instead using our own configuration
    config = Config.default()
    msg_format = config.log_format

    initialise_logging(level=loglevel, format=msg_format, log_directory=config.paths.log)

worker_ready_handler(sender=None, **kwargs) #

Log a message when the worker is ready

Source code in packages/climate-ref-celery/src/climate_ref_celery/app.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@worker_ready.connect
def worker_ready_handler(sender: Any = None, **kwargs: Any) -> None:  # pragma: no cover
    """
    Log a message when the worker is ready
    """
    config = Config.default()
    logger.info(f"Worker ready with config {pretty_repr(config)}")

    max_tasks = sender.app.conf.worker_max_tasks_per_child if sender is not None else None
    if config.executor.measure_resources and max_tasks != 1:
        # The bias is not recorded on the row, so it has to be said out loud here.
        logger.warning(
            "CELERY_WORKER_MAX_TASKS_PER_CHILD is not 1, "
            "so worker processes are reused across executions. "
            "A recorded peak memory can include the footprint of an earlier execution "
            "in the same worker."
        )