Add hooks

Author: T4rk1n

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,83 @@
+from flask import jsonify
+import requests
+
+from dash import Dash, Input, Output, html, hooks, set_props
+
+
+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")