Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add REPL context #3042

Merged
merged 8 commits into from
Mar 5, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions sanic/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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()
Expand Down
51 changes: 46 additions & 5 deletions sanic/cli/console.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down
69 changes: 69 additions & 0 deletions sanic/models/ctx_types.py
Original file line number Diff line number Diff line change
@@ -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
48 changes: 48 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
)
Loading