I was trying to diagnose a complex interaction across several callbacks. Is there an inbuilt functionality to log which callbacks fire?
I couldn’t find anything, so I created a decorator for logging callback context when a callback fires (callback_telemetry
), and I included it in a custom class that extends Dash (DashWithTelemetry
).
Then I created my app using DashWithTelemetry
instead of dash.Dash
, and I was off to the races.
""" callback_telemetry.py -- Extend Dash to include logging of callbacks w/ context """
import inspect
import logging
import time
from functools import wraps
from dash import Dash, callback_context
LOG_CALLBACK_TELEMETRY = True
log = logging.getLogger(__name__)
def callback_telemetry(func):
"""wrapper to provide telemetry for dash callbacks"""
@wraps(func)
def timeit_wrapper(*args, **kwargs):
def get_callback_ref(func_ref):
module = inspect.getmodule(func_ref)
return f"{module.__name__.split('.')[-1]}:{func_ref.__name__}"
def generate_results_dict(function_output, outputs_list):
if isinstance(function_output, tuple):
output_strs = [
f"{output['id']}.{output['property']}" for output in outputs_list
]
return dict(zip(output_strs, function_output))
return {f"{outputs_list['id']}.{outputs_list['property']}": function_output}
def format_callback_dict(data):
return "||".join([f"{key}:{str(data[key])[:20]}" for key in data])
start_time = time.perf_counter()
result = func(*args, **kwargs)
end_time = time.perf_counter()
total_time = end_time - start_time
results_dict = generate_results_dict(result, callback_context.outputs_list)
inputs_str = format_callback_dict(callback_context.inputs)
state_str = format_callback_dict(callback_context.states)
result_str = format_callback_dict(results_dict)
context = (
f"___input:|{inputs_str}|\n___state:|{state_str}|\n__output:|{result_str}|"
)
log.info(f"[TELEMETRY] {get_callback_ref(func)}, {total_time:.4f}s\n{context}")
return result
return timeit_wrapper
class DashWithTelemetry(Dash):
"""Provide logging telemetry for Dash callbacks"""
def callback(self, *_args, **_kwargs):
def decorator(function):
@super(DashWithTelemetry, self).callback( # pylint: disable=R1725
*_args, **_kwargs
)
def wrapper(*args, **kwargs):
if LOG_CALLBACK_TELEMETRY:
retval = (callback_telemetry)(function)(*args, **kwargs)
else:
retval = function(*args, **kwargs)
return retval
return wrapper
return decorator