Function _get_callback_manager

dash/_callback.py:389–414

Set up the background callback and manage jobs.

(
    kwargs: dict, background: dict
)

def _get_callback_manager(
    kwargs: dict, background: dict
) -> BaseBackgroundCallbackManager:
    """Set up the background callback and manage jobs."""
    callback_manager = background.get(
        "manager", kwargs.get("background_callback_manager", None)
    )
    if background is not None:
        if not callback_manager:
            raise MissingLongCallbackManagerError(
                "Running `background` callbacks requires a manager to be installed.\n"
                "Available managers:\n"
                "- Diskcache (`pip install dash[diskcache]`) to run callbacks in a separate Process"
                " and store results on the local filesystem.\n"
                "- Celery (`pip install dash[celery]`) to run callbacks in a celery worker"
                " and store results on redis.\n"
            )

    adapter = get_app().backend.request_adapter()
    old_job = adapter.args.getlist("oldJob") if hasattr(adapter.args, "getlist") else []

    if old_job:
        for job in old_job:
            callback_manager.terminate_job(job)

    return callback_manager


def _setup_background_callback(