Skip to content
Draft
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
8 changes: 8 additions & 0 deletions litestar/cli/_shell_startup.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
import os

from litestar.cli._utils import _load_app_from_path, populate_repl_globals

app = _load_app_from_path(os.environ["LITESTAR_APP"]).app
new_locals, banner = populate_repl_globals(app=app)
print(banner)
globals().update(new_locals)
13 changes: 13 additions & 0 deletions litestar/cli/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -581,3 +581,16 @@ def isatty() -> bool:
This is a convenience wrapper around the built in system methods. This allows for easier testing of TTY/non-TTY modes.
"""
return sys.stdout.isatty()


def populate_repl_globals(app: Litestar) -> tuple[dict[str, Any], str]:
repl_locals = {"app": Litestar}
banner = "app = Litestar(...)\n\n"
for plugin in app.plugins.cli:
plugin_provided = plugin.populate_repl_namespace(app=app)
if plugin_provided:
banner += f"Plugin {type(plugin).__name__!r} provided:\n"
for name, value in plugin_provided.items():
repl_locals[name] = value
banner += f"name = {value!r}\n"
return repl_locals, banner
87 changes: 86 additions & 1 deletion litestar/cli/commands/core.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
from __future__ import annotations

import importlib.resources
import inspect
import multiprocessing
import os
import subprocess
import sys
from contextlib import AbstractContextManager, ExitStack, contextmanager
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal

try:
import rich_click as click
Expand All @@ -21,6 +22,7 @@
console,
create_ssl_files,
isatty,
populate_repl_globals,
remove_default_schema_routes,
remove_routes_with_patterns,
show_app_info,
Expand Down Expand Up @@ -365,3 +367,86 @@ def _handle_http_route(self, route: HTTPRoute) -> None:
branch.add(" ".join([f"[green]{path}[green]", *handler_info]))
else:
branch.add(" ".join(handler_info))


def _autoselect_repl_module() -> Literal["repl", "asyncio", "ipython"]:
import importlib.util

if importlib.util.find_spec("IPython"):
return "ipython"

if sys.version_info >= (3, 13):
return "asyncio"

return "repl"


@click.command(name="shell")
@click.option(
"--repl",
default=None,
type=click.Choice(("repl", "asyncio", "ipython")),
envvar="LITESTAR_REPL",
required=False,
help="Start a Python shell with the Litestar application and other values provided by plugins preloaded",
)
def shell_command(
app: Litestar,
repl: Literal["repl", "asyncio", "ipython"] | None = None,
) -> None: # pragma: no cover
if repl is None:
repl = _autoselect_repl_module()
click.secho(f"Starting Litestar shell using autoselected REPL {repl!r}", fg="blue")

if repl == "asyncio":
if sys.version_info < (3, 13):
click.secho("Litestar shell using the asyncio REPL requires Python 3.13 or greater", fg="red")
click.secho(
"To use the Litestar shell with an async REPL in this version of Python, y"
"ou can install Litestar with the 'ipython' extra (litestar[ipython]) "
"and then select ipython as the repl, either by starting he Litestar "
"shell with 'litestar shell --repl=ipython' or setting the "
"'LITESTAR_REPL=ipython' environment variable",
fg="blue",
)
quit(1)

# since it's currently not possible to customise e.g. the namespace of the
# asyncio REPL, we have to get a bit creative here

if "PYTHONSTARTUP" in os.environ: # type: ignore[unreachable]
click.secho(
"Cannot run Litestar shell with asyncio REPL when PYTHONSTARTUP is "
"set. PYTHONSTARTUP is currently set to "
f"{os.environ['PYTHONSTARTUP']!r}. Either unset PYTHONSTARTUP or use a "
"different REPL ('repl' or 'ipython').",
fg="red",
)

subprocess.run( # noqa: S603
[sys.executable, "-m", "asyncio"],
check=False,
env={
**os.environ,
"PYTHONSTARTUP": str(importlib.resources.files("litestar.cli").joinpath("_shell_startup")) + ".py",
},
)
elif repl == "repl":
import code

repl_locals, banner = populate_repl_globals(app=app)
interpreter = code.InteractiveConsole(locals=repl_locals)
interpreter.interact(banner=banner)
elif repl == "ipython":
import IPython
from traitlets.config.loader import Config

repl_locals, banner = populate_repl_globals(app=app)

config = Config()
config.TerminalInteractiveShell.banner2 = banner

IPython.start_ipython(argv=[], user_ns=repl_locals, config=config) # type: ignore[no-untyped-call]
else:
click.secho(f"Unsupported REPL {repl!r}") # type: ignore[unreachable]
quit(1)
1 change: 1 addition & 0 deletions litestar/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,4 @@ def litestar_group(ctx: click.Context, app_path: str | None, app_dir: Path | Non
litestar_group.add_command(core.version_command) # pyright: ignore
litestar_group.add_command(sessions.sessions_group) # pyright: ignore
litestar_group.add_command(schema.schema_group) # pyright: ignore
litestar_group.add_command(core.shell_command) # pyright: ignore
6 changes: 6 additions & 0 deletions litestar/plugins/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,12 @@ def is_debug_mode(app: Litestar):
def server_lifespan(self, app: Litestar) -> Iterator[None]:
yield

def populate_repl_namespace(self, app: Litestar) -> dict[str, Any]:
"""Return a dict that will be used to populate the REPL namespace of the
Litestar shell.
"""
return {}


class SerializationPlugin(abc.ABC):
"""Abstract base class for plugins that extend DTO functionality"""
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,7 @@ dev = [
"daphne>=4.0.0",
"opentelemetry-sdk",
"httpx-sse",
"ipython",
]

docs = [
Expand Down
Loading
Loading