Function _setup_background_callback

dash/_callback.py:417–466

Set up the background callback and manage jobs.

(
    kwargs, background, background_key, func, func_args, func_kwargs, callback_ctx
)

def _setup_background_callback(
    kwargs, background, background_key, func, func_args, func_kwargs, callback_ctx
):
    """Set up the background callback and manage jobs."""
    callback_manager = _get_callback_manager(kwargs, background)
    if not callback_manager:
        return to_json({"error": "No background callback manager configured"})

    progress_outputs = background.get("progress")

    cache_ignore_triggered = background.get("cache_ignore_triggered", True)

    cache_key = callback_manager.build_cache_key(
        func,
        func_args if func_args else func_kwargs,
        background.get("cache_args_to_ignore", []),
        None if cache_ignore_triggered else callback_ctx.get("triggered_inputs", []),
    )
    job_fn = callback_manager.func_registry.get(background_key)
    ctx_value = AttributeDict(**context_value.get())
    ctx_value.ignore_register_page = True
    ctx_value.pop("background_callback_manager")
    ctx_value.pop("dash_response")

    args_value = ctx_value.get("args")
    if args_value is not None and not isinstance(args_value, dict):
        ctx_value["args"] = dict(args_value)

    job = callback_manager.call_job_fn(
        cache_key,
        job_fn,
        func_args if func_args else func_kwargs,
        ctx_value,
    )

    data = {
        "cacheKey": cache_key,
        "job": job,
    }

    cancel = background.get("cancel")
    if cancel:
        data["cancel"] = cancel

    progress_default = background.get("progressDefault")
    if progress_default:
        data["progressDefault"] = {
            str(o): x for o, x in zip(progress_outputs, progress_default)
        }
    return to_json(data)


def _progress_background_callback(