Add hooks

dash/__init__.py

@@ -41,6 +41,8 @@
 from ._patch import Patch # noqa: F401,E402
 from ._jupyter import jupyter_dash # noqa: F401,E402
 
+from . import _hooks as hooks # noqa: F401,E402
+
 ctx = callback_context
 
 

dash/_hooks.py

@@ -0,0 +1,93 @@
+import typing as _t
+
+_ns = {
+ "setup": [],
+ "layout": [],
+ "routes": [],
+ "error": [],
+ "callback": [],
+}
+
+
+def layout(func):
+ """
+ Run a function when serving the layout, the return value
+ will be used as the layout.
+ """
+ _ns["layout"].append(func)
+ return func
+
+
+def setup(func):
+ """
+ Can be used to get a reference to the app after it is instantiated.
+ """
+ _ns["setup"].append(func)
+ return func
+
+
+def route(name=None, methods=("GET",)):
+ """
+ Add a route to the Dash server.
+ """
+
+ def wrap(func):
+ _name = name or func.__name__
+ _ns["routes"].append((_name, func, methods))
+ return func
+
+ return wrap
+
+
+def error(func: _t.Callable[[Exception], _t.Any]):
+ """Automatically add an error handler to the dash app."""
+ _ns["error"].append(func)
+ return func
+
+
+def callback(*args, **kwargs):
+ """
+ Add a callback to all the apps with the hook installed.
+ """
+
+ def wrap(func):
+ _ns["callback"].append((list(args), dict(kwargs), func))
+ return func
+
+ return wrap
+
+
+class HooksManager:
+ _registered = False
+
+ # pylint: disable=too-few-public-methods
+ class HookErrorHandler:
+ def __init__(self, original, hooks):
+ self.original = original
+ self.hooks = hooks
+
+ def __call__(self, err: Exception):
+ result = None
+ if self.original:
+ result = self.original(err)
+ hook_result = None
+ for hook in HooksManager.get_hooks("error"):
+ hook_result = hook(err)
+ return result or hook_result
+
+ @staticmethod
+ def get_hooks(hook: str):
+ return _ns.get(hook, []).copy()
+
+ @classmethod
+ def register_setuptools(cls):
+ if cls._registered:
+ return
+
+ import importlib.metadata # pylint: disable=import-outside-toplevel
+
+ for dist in importlib.metadata.distributions():
+ for entry in dist.entry_points:
+ if entry.group != "dash":
+ continue
+ entry.load()

dash/dash.py

@@ -573,6 +573,8 @@ def __init__( # pylint: disable=too-many-statements
  for plugin in plugins:
  plugin.plug(self)
 
+ self._setup_hooks()
+
  # tracks internally if a function already handled at least one request.
  self._got_first_request = {"pages": False, "setup_server": False}
 
@@ -588,6 +590,25 @@ def __init__( # pylint: disable=too-many-statements
  )
  self.setup_startup_routes()
 
+ def _setup_hooks(self):
+ # pylint: disable=import-outside-toplevel
+ from ._hooks import HooksManager
+
+ self._hooks = HooksManager
+ self._hooks.register_setuptools()
+
+ for setup in self._hooks.get_hooks("setup"):
+ setup(self)
+
+ for callback_args, callback_kwargs, callback in self._hooks.get_hooks(
+ "callback"
+ ):
+ self.callback(*callback_args, **callback_kwargs)(callback)
+
+ error_handler = self._hooks.get_hooks("error")
+ if error_handler:
+ self._on_error = self._hooks.HookErrorHandler(self._on_error, error_handler)
+
  def init_app(self, app=None, **kwargs):
  """Initialize the parts of Dash that require a flask app."""
 
@@ -688,6 +709,9 @@ def _setup_routes(self):
  "_alive_" + jupyter_dash.alive_token, jupyter_dash.serve_alive
  )
 
+ for name, func, methods in self._hooks.get_hooks("routes"):
+ self._add_url(name, func, methods)
+
  # catch-all for front-end routes, used by dcc.Location
  self._add_url("<path:path>", self.index)
 
@@ -754,6 +778,9 @@ def index_string(self, value):
  def serve_layout(self):
  layout = self._layout_value()
 
+ for hook in self._hooks.get_hooks("layout"):
+ layout = hook(layout)
+
  # TODO - Set browser cache limit - pass hash into frontend
  return flask.Response(
  to_json(layout),

tests/integration/test_hooks.py

@@ -0,0 +1,93 @@
+from flask import jsonify
+import requests
+import pytest
+
+from dash import Dash, Input, Output, html, hooks, set_props
+
+
+@pytest.fixture(scope="module", autouse=True)
+def hook_cleanup():
+ hooks._ns["layout"] = []
+ hooks._ns["setup"] = []
+ hooks._ns["route"] = []
+ hooks._ns["error"] = []
+ hooks._ns["callback"] = []
+
+
+def test_hook001_layout(dash_duo):
+ @hooks.layout
+ def on_layout(layout):
+ return [html.Div("Header", id="header")] + layout
+
+ app = Dash()
+ app.layout = [html.Div("Body", id="body")]
+
+ dash_duo.start_server(app)
+
+ dash_duo.wait_for_text_to_equal("#header", "Header")
+ dash_duo.wait_for_text_to_equal("#body", "Body")
+
+
+def test_hook002_setup():
+ setup_title = None
+
+ @hooks.setup
+ def on_setup(app: Dash):
+ nonlocal setup_title
+ setup_title = app.title
+
+ app = Dash(title="setup-test")
+ app.layout = html.Div("setup")
+
+ assert setup_title == "setup-test"
+
+
+def test_hook003_route(dash_duo):
+ @hooks.route(methods=("POST",))
+ def hook_route():
+ return jsonify({"success": True})
+
+ app = Dash()
+ app.layout = html.Div("hook route")
+
+ dash_duo.start_server(app)
+ response = requests.post(f"{dash_duo.server_url}/hook_route")
+ data = response.json()
+ assert data["success"]
+
+
+def test_hook004_error(dash_duo):
+ @hooks.error
+ def on_error(error):
+ set_props("error", {"children": str(error)})
+
+ app = Dash()
+ app.layout = [html.Button("start", id="start"), html.Div(id="error")]
+
+ @app.callback(Input("start", "n_clicks"), prevent_initial_call=True)
+ def on_click(_):
+ raise Exception("hook error")
+
+ dash_duo.start_server(app)
+ dash_duo.wait_for_element("#start").click()
+ dash_duo.wait_for_text_to_equal("#error", "hook error")
+
+
+def test_hook005_callback(dash_duo):
+ @hooks.callback(
+ Output("output", "children"),
+ Input("start", "n_clicks"),
+ prevent_initial_call=True,
+ )
+ def on_hook_cb(n_clicks):
+ return f"clicked {n_clicks}"
+
+ app = Dash()
+ app.layout = [
+ html.Button("start", id="start"),
+ html.Div(id="output"),
+ ]
+
+ dash_duo.start_server(app)
+ dash_duo.wait_for_element("#start").click()
+ dash_duo.wait_for_text_to_equal("#output", "clicked 1")