Add REPL context (#3042)

* Add REPL context

* fix(type): add missing return type

* chore: rename method and add docstring

* fix: handle null docstring

* chore: add test case

* Add direct attr setting

---------

Co-authored-by: goodki-d <185676088+goodki-d@users.noreply.github.com>

Author: ahopkins

sanic/app.py

@@ -70,6 +70,7 @@
 from sanic.mixins.listeners import ListenerEvent
 from sanic.mixins.startup import StartupMixin
 from sanic.mixins.static import StaticHandleMixin
+from sanic.models.ctx_types import REPLContext
 from sanic.models.futures import (
     FutureException,
     FutureListener,
@@ -209,6 +210,7 @@ class to use for the application. Defaults to `None`.
         "multiplexer",
         "named_request_middleware",
         "named_response_middleware",
+        "repl_ctx",
         "request_class",
         "request_middleware",
         "response_middleware",
@@ -372,6 +374,7 @@ def __init__(
         self.listeners: dict[str, list[ListenerType[Any]]] = defaultdict(list)
         self.named_request_middleware: dict[str, Deque[Middleware]] = {}
         self.named_response_middleware: dict[str, Deque[Middleware]] = {}
+        self.repl_ctx: REPLContext = REPLContext()
         self.request_class = request_class or Request
         self.request_middleware: Deque[Middleware] = deque()
         self.response_middleware: Deque[Middleware] = deque()

sanic/cli/console.py

@@ -18,6 +18,7 @@
 from sanic.helpers import Default
 from sanic.http.constants import Stage
 from sanic.log import Colors
+from sanic.models.ctx_types import REPLContext
 from sanic.models.protocol_types import TransportProtocol
 from sanic.response.types import HTTPResponse
 
@@ -110,6 +111,13 @@ async def do(
     return Result(request, response)
 
 
+def _variable_description(name: str, desc: str, type_desc: str) -> str:
+    return (
+        f"  - {Colors.BOLD + Colors.SANIC}{name}{Colors.END}: {desc} - "
+        f"{Colors.BOLD + Colors.BLUE}{type_desc}{Colors.END}"
+    )
+
+
 class SanicREPL(InteractiveConsole):
     def __init__(self, app: Sanic, start: Optional[Default] = None):
         global repl_app
@@ -119,24 +127,47 @@ def __init__(self, app: Sanic, start: Optional[Default] = None):
             "sanic": sanic,
             "do": do,
         }
+
+        user_locals = {
+            user_local.name: user_local.var for user_local in app.repl_ctx
+        }
+
         client_availability = ""
         variable_descriptions = [
-            f"  - {Colors.BOLD + Colors.SANIC}app{Colors.END}: The Sanic application instance - {Colors.BOLD + Colors.BLUE}{str(app)}{Colors.END}",  # noqa: E501
-            f"  - {Colors.BOLD + Colors.SANIC}sanic{Colors.END}: The Sanic module - {Colors.BOLD + Colors.BLUE}import sanic{Colors.END}",  # noqa: E501
-            f"  - {Colors.BOLD + Colors.SANIC}do{Colors.END}: An async function to fake a request to the application - {Colors.BOLD + Colors.BLUE}Result(request, response){Colors.END}",  # noqa: E501
+            _variable_description(
+                "app", REPLContext.BUILTINS["app"], str(app)
+            ),
+            _variable_description(
+                "sanic", REPLContext.BUILTINS["sanic"], "import sanic"
+            ),
+            _variable_description(
+                "do", REPLContext.BUILTINS["do"], "Result(request, response)"
+            ),
         ]
+
+        user_locals_descriptions = [
+            _variable_description(
+                user_local.name, user_local.desc, str(type(user_local.var))
+            )
+            for user_local in app.repl_ctx
+        ]
+
         if HTTPX_AVAILABLE:
             locals_available["client"] = SanicClient(app)
             variable_descriptions.append(
-                f"  - {Colors.BOLD + Colors.SANIC}client{Colors.END}: A client to access the Sanic app instance using httpx - {Colors.BOLD + Colors.BLUE}from httpx import Client{Colors.END}",  # noqa: E501
+                _variable_description(
+                    "client",
+                    REPLContext.BUILTINS["client"],
+                    "from httpx import Client",
+                ),
             )
         else:
             client_availability = (
                 f"\n{Colors.YELLOW}The HTTP client has been disabled. "
                 "To enable it, install httpx:\n\t"
                 f"pip install httpx{Colors.END}\n"
             )
-        super().__init__(locals=locals_available)
+        super().__init__(locals={**locals_available, **user_locals})
         self.compile.compiler.flags |= PyCF_ALLOW_TOP_LEVEL_AWAIT
         self.loop = new_event_loop()
         self._start = start
@@ -163,6 +194,16 @@ def __init__(self, app: Sanic, start: Optional[Default] = None):
                 client_availability,
                 "The following objects are available for your convenience:",  # noqa: E501
                 *variable_descriptions,
+            ]
+            + (
+                [
+                    "\nREPL Context:",
+                    *user_locals_descriptions,
+                ]
+                if user_locals_descriptions
+                else []
+            )
+            + [
                 "\nThe async/await keywords are available for use here.",  # noqa: E501
                 f"To exit, press {Colors.BOLD}CTRL+C{Colors.END}, "
                 f"{Colors.BOLD}CTRL+D{Colors.END}, or type {Colors.BOLD}exit(){Colors.END}.\n",  # noqa: E501

sanic/models/ctx_types.py

@@ -0,0 +1,69 @@
+from typing import Any, NamedTuple, Optional
+
+
+class REPLLocal(NamedTuple):
+    var: Any
+    name: str
+    desc: str
+
+
+class REPLContext:
+    BUILTINS = {
+        "app": "The Sanic application instance",
+        "sanic": "The Sanic module",
+        "do": "An async function to fake a request to the application",
+        "client": "A client to access the Sanic app instance using httpx",
+    }
+
+    def __init__(self):
+        self._locals: set[REPLLocal] = set()
+
+    def add(
+        self,
+        var: Any,
+        name: Optional[str] = None,
+        desc: Optional[str] = None,
+    ):
+        """Add a local variable to be available in REPL context.
+
+        Args:
+            var (Any): A module, class, object or a class.
+            name (Optional[str], optional): An alias for the local. Defaults to None.
+            desc (Optional[str], optional): A brief description for the local. Defaults to None.
+        """  # noqa: E501
+        if name is None:
+            try:
+                name = var.__name__
+            except AttributeError:
+                name = var.__class__.__name__
+
+        if desc is None:
+            try:
+                desc = var.__doc__ or ""
+            except AttributeError:
+                desc = str(type(var))
+
+        assert isinstance(desc, str) and isinstance(
+            name, str
+        )  # Just to make mypy happy
+
+        if name in self.BUILTINS:
+            raise ValueError(f"Cannot override built-in variable: {name}")
+
+        desc = self._truncate(desc)
+
+        self._locals.add(REPLLocal(var, name, desc))
+
+    def __setattr__(self, name: str, value: Any):
+        if name.startswith("_"):
+            super().__setattr__(name, value)
+        else:
+            self.add(value, name)
+
+    def __iter__(self):
+        return iter(self._locals)
+
+    @staticmethod
+    def _truncate(s: str, limit: int = 40) -> str:
+        s = s.replace("\n", " ")
+        return s[:limit] + "..." if len(s) > limit else s

tests/test_cli.py

@@ -13,6 +13,7 @@
 from sanic import __version__
 from sanic.__main__ import main
 from sanic.cli.inspector_client import InspectorClient
+from sanic.models.ctx_types import REPLLocal
 
 from .conftest import get_port
 
@@ -404,3 +405,50 @@ def test_command_with_renamed_command(caplog):
     with patch("sys.argv", ["sanic", *args]):
         lines = capture(args, caplog)
     assert "BAZ" in lines
+
+
+def test_add_local_method(app):
+    def foo(): ...
+    def bar():
+        """bar method docstring."""
+
+    class Luffy: ...
+
+    import os
+
+    app.repl_ctx.add(foo)
+    app.repl_ctx.add(bar)
+    app.repl_ctx.add(Luffy)
+    app.repl_ctx.add(os, desc="Standard os module.")
+
+    assert REPLLocal(foo, "foo", "") in app.repl_ctx._locals
+    assert (
+        REPLLocal(bar, "bar", "bar method docstring.") in app.repl_ctx._locals
+    )
+    assert REPLLocal(Luffy, "Luffy", "") in app.repl_ctx._locals
+    assert REPLLocal(os, "os", "Standard os module.") in app.repl_ctx._locals
+
+
+def test_add_local_attr(app):
+    def foo(): ...
+    def bar():
+        """bar method docstring."""
+
+    class Luffy: ...
+
+    import os
+
+    app.repl_ctx.foo = foo
+    app.repl_ctx.bar = bar
+    app.repl_ctx.Luffy = Luffy
+    app.repl_ctx.os = os
+
+    assert REPLLocal(foo, "foo", "") in app.repl_ctx._locals
+    assert (
+        REPLLocal(bar, "bar", "bar method docstring.") in app.repl_ctx._locals
+    )
+    assert REPLLocal(Luffy, "Luffy", "") in app.repl_ctx._locals
+    assert any(
+        isinstance(item, REPLLocal) and item.var == os and item.name == "os"
+        for item in app.repl_ctx._locals
+    )