Source code for visualization_toolkit.helpers.dash.celery

from celery import Celery
from dash import CeleryManager




def setup_celery_app(
    name: str,
    broker_uri: str,
    results_backend_uri: str = None,
    celery_app_kwargs: dict = None,
    celery_manager_kwargs: dict = None,
) -> (Celery, CeleryManager):
    """
    Intialize a celery application which is needed if using background callbacks with Dash.
    Generally this is a preferred setup for production dash applications.

    Examples
    ^^^^^^^^^^^^^
    .. code-block:: python
        :caption: Example of setting up a celery app with background callback caching

        import uuid
        from visualization_toolkit.helpers.dash import setup_celery_app
        from visualization_toolkit.constants import APP_ENVIRONMENT

        # Use a different cache_by logic based on the needs of the dash application
        launch_uid = str(uuid.uuid4())
        celery_app, celery_app_manager = setup_celery_app(
            name=__name__,
            broker_uri=APP_ENVIRONMENT.get("REDIS_URI"),
            celery_manager_kwargs={
                "cache_by": [lambda: launch_uid],
            },
        )


    :param name: Celery application name
    :param broker_uri: Redis URI to use as the celery broker
    :param results_backend_uri: Optional URI to use as the celery results backend. This will default to ``broker_uri`` if not set. Either redis or postgres can be used.
    :param celery_app_kwargs: Additional keyword arguments to pass to the celery app.
    :param celery_manager_kwargs: Additional keyword arguments to pass to the celery manager.
    :return: Tuple containing the celery app and the cellery callback manager. Both of which are used in backgorund callbacks in Dash.
    """
    results_backend_uri = results_backend_uri or broker_uri
    celery_app_kwargs = celery_app_kwargs or {}
    celery_manager_kwargs = celery_manager_kwargs or {}

    celery_app = Celery(
        name, broker=broker_uri, backend=results_backend_uri, **celery_app_kwargs
    )
    background_callback_manager = CeleryManager(celery_app, **celery_manager_kwargs)

    return celery_app, background_callback_manager