From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../site-packages/litestar/cli/__init__.py | 29 ++ .../cli/__pycache__/__init__.cpython-311.pyc | Bin 0 -> 1276 bytes .../cli/__pycache__/_utils.cpython-311.pyc | Bin 0 -> 34239 bytes .../litestar/cli/__pycache__/main.cpython-311.pyc | Bin 0 -> 2459 bytes .../site-packages/litestar/cli/_utils.py | 562 +++++++++++++++++++++ .../litestar/cli/commands/__init__.py | 0 .../commands/__pycache__/__init__.cpython-311.pyc | Bin 0 -> 205 bytes .../cli/commands/__pycache__/core.cpython-311.pyc | Bin 0 -> 16664 bytes .../commands/__pycache__/schema.cpython-311.pyc | Bin 0 -> 4442 bytes .../commands/__pycache__/sessions.cpython-311.pyc | Bin 0 -> 3813 bytes .../site-packages/litestar/cli/commands/core.py | 335 ++++++++++++ .../site-packages/litestar/cli/commands/schema.py | 82 +++ .../litestar/cli/commands/sessions.py | 58 +++ .../python3.11/site-packages/litestar/cli/main.py | 37 ++ 14 files changed, 1103 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/__init__.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/__pycache__/__init__.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/__pycache__/_utils.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/__pycache__/main.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/_utils.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/__init__.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/__init__.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/core.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/schema.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/sessions.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/core.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/schema.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/main.py (limited to 'venv/lib/python3.11/site-packages/litestar/cli') diff --git a/venv/lib/python3.11/site-packages/litestar/cli/__init__.py b/venv/lib/python3.11/site-packages/litestar/cli/__init__.py new file mode 100644 index 0000000..f6c366e --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/__init__.py @@ -0,0 +1,29 @@ +"""Litestar CLI.""" + +from __future__ import annotations + +from importlib.util import find_spec + +# Ensure `rich_click` patching occurs before we do any imports from `click`. +if find_spec("rich_click") is not None: # pragma: no cover + import rich_click as click + from rich_click.cli import patch as rich_click_patch + + rich_click_patch() + click.rich_click.USE_RICH_MARKUP = True + click.rich_click.USE_MARKDOWN = False + click.rich_click.SHOW_ARGUMENTS = True + click.rich_click.GROUP_ARGUMENTS_OPTIONS = True + click.rich_click.SHOW_ARGUMENTS = True + click.rich_click.GROUP_ARGUMENTS_OPTIONS = True + click.rich_click.STYLE_ERRORS_SUGGESTION = "magenta italic" + click.rich_click.ERRORS_SUGGESTION = "" + click.rich_click.ERRORS_EPILOGUE = "" + click.rich_click.MAX_WIDTH = 80 + click.rich_click.SHOW_METAVARS_COLUMN = True + click.rich_click.APPEND_METAVARS_HELP = True + + +from .main import litestar_group + +__all__ = ["litestar_group"] diff --git a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..808a2fb Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/__init__.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/_utils.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/_utils.cpython-311.pyc new file mode 100644 index 0000000..4baba11 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/_utils.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/main.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/main.cpython-311.pyc new file mode 100644 index 0000000..7ede592 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/main.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/_utils.py b/venv/lib/python3.11/site-packages/litestar/cli/_utils.py new file mode 100644 index 0000000..f36cd77 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/_utils.py @@ -0,0 +1,562 @@ +from __future__ import annotations + +import contextlib +import importlib +import inspect +import os +import re +import sys +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from functools import wraps +from importlib.util import find_spec +from itertools import chain +from os import getenv +from pathlib import Path +from typing import TYPE_CHECKING, Any, Callable, Generator, Iterable, Sequence, TypeVar, cast + +from click import ClickException, Command, Context, Group, pass_context +from rich import get_console +from rich.table import Table +from typing_extensions import ParamSpec, get_type_hints + +from litestar import Litestar, __version__ +from litestar.middleware import DefineMiddleware +from litestar.utils import get_name + +if sys.version_info >= (3, 10): + from importlib.metadata import entry_points +else: + from importlib_metadata import entry_points + + +if TYPE_CHECKING: + from litestar.openapi import OpenAPIConfig + from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute + from litestar.types import AnyCallable + + +UVICORN_INSTALLED = find_spec("uvicorn") is not None +JSBEAUTIFIER_INSTALLED = find_spec("jsbeautifier") is not None + + +__all__ = ( + "UVICORN_INSTALLED", + "JSBEAUTIFIER_INSTALLED", + "LoadedApp", + "LitestarCLIException", + "LitestarEnv", + "LitestarExtensionGroup", + "LitestarGroup", + "show_app_info", +) + + +P = ParamSpec("P") +T = TypeVar("T") + + +AUTODISCOVERY_FILE_NAMES = ["app", "application"] + +console = get_console() + + +class LitestarCLIException(ClickException): + """Base class for Litestar CLI exceptions.""" + + def __init__(self, message: str) -> None: + """Initialize exception and style error message.""" + super().__init__(message) + + +@dataclass +class LitestarEnv: + """Information about the current Litestar environment variables.""" + + app_path: str + debug: bool + app: Litestar + cwd: Path + host: str | None = None + port: int | None = None + fd: int | None = None + uds: str | None = None + reload: bool | None = None + reload_dirs: tuple[str, ...] | None = None + reload_include: tuple[str, ...] | None = None + reload_exclude: tuple[str, ...] | None = None + web_concurrency: int | None = None + is_app_factory: bool = False + certfile_path: str | None = None + keyfile_path: str | None = None + create_self_signed_cert: bool = False + + @classmethod + def from_env(cls, app_path: str | None, app_dir: Path | None = None) -> LitestarEnv: + """Load environment variables. + + If ``python-dotenv`` is installed, use it to populate environment first + """ + cwd = Path().cwd() if app_dir is None else app_dir + cwd_str_path = str(cwd) + if cwd_str_path not in sys.path: + sys.path.append(cwd_str_path) + + with contextlib.suppress(ImportError): + import dotenv + + dotenv.load_dotenv() + app_path = app_path or getenv("LITESTAR_APP") + if app_path and getenv("LITESTAR_APP") is None: + os.environ["LITESTAR_APP"] = app_path + if app_path: + console.print(f"Using Litestar app from env: [bright_blue]{app_path!r}") + loaded_app = _load_app_from_path(app_path) + else: + loaded_app = _autodiscover_app(cwd) + + port = getenv("LITESTAR_PORT") + web_concurrency = getenv("WEB_CONCURRENCY") + uds = getenv("LITESTAR_UNIX_DOMAIN_SOCKET") + fd = getenv("LITESTAR_FILE_DESCRIPTOR") + reload_dirs = tuple(s.strip() for s in getenv("LITESTAR_RELOAD_DIRS", "").split(",") if s) or None + reload_include = tuple(s.strip() for s in getenv("LITESTAR_RELOAD_INCLUDES", "").split(",") if s) or None + reload_exclude = tuple(s.strip() for s in getenv("LITESTAR_RELOAD_EXCLUDES", "").split(",") if s) or None + + return cls( + app_path=loaded_app.app_path, + app=loaded_app.app, + debug=_bool_from_env("LITESTAR_DEBUG"), + host=getenv("LITESTAR_HOST"), + port=int(port) if port else None, + uds=uds, + fd=int(fd) if fd else None, + reload=_bool_from_env("LITESTAR_RELOAD"), + reload_dirs=reload_dirs, + reload_include=reload_include, + reload_exclude=reload_exclude, + web_concurrency=int(web_concurrency) if web_concurrency else None, + is_app_factory=loaded_app.is_factory, + cwd=cwd, + certfile_path=getenv("LITESTAR_SSL_CERT_PATH"), + keyfile_path=getenv("LITESTAR_SSL_KEY_PATH"), + create_self_signed_cert=_bool_from_env("LITESTAR_CREATE_SELF_SIGNED_CERT"), + ) + + +@dataclass +class LoadedApp: + """Information about a loaded Litestar app.""" + + app: Litestar + app_path: str + is_factory: bool + + +class LitestarGroup(Group): + """:class:`click.Group` subclass that automatically injects ``app`` and ``env` kwargs into commands that request it. + + Use this as the ``cls`` for :class:`click.Group` if you're extending the internal CLI with a group. For ``command``s + added directly to the root group this is not needed. + """ + + def __init__( + self, + name: str | None = None, + commands: dict[str, Command] | Sequence[Command] | None = None, + **attrs: Any, + ) -> None: + """Init ``LitestarGroup``""" + self.group_class = LitestarGroup + super().__init__(name=name, commands=commands, **attrs) + + def add_command(self, cmd: Command, name: str | None = None) -> None: + """Add command. + + If necessary, inject ``app`` and ``env`` kwargs + """ + if cmd.callback: + cmd.callback = _inject_args(cmd.callback) + super().add_command(cmd) + + def command(self, *args: Any, **kwargs: Any) -> Callable[[AnyCallable], Command] | Command: # type: ignore[override] + # For some reason, even when copying the overloads + signature from click 1:1, mypy goes haywire + """Add a function as a command. + + If necessary, inject ``app`` and ``env`` kwargs + """ + + def decorator(f: AnyCallable) -> Command: + f = _inject_args(f) + return cast("Command", Group.command(self, *args, **kwargs)(f)) + + return decorator + + +class LitestarExtensionGroup(LitestarGroup): + """``LitestarGroup`` subclass that will load Litestar-CLI extensions from the `litestar.commands` entry_point. + + This group class should not be used on any group besides the root ``litestar_group``. + """ + + def __init__( + self, + name: str | None = None, + commands: dict[str, Command] | Sequence[Command] | None = None, + **attrs: Any, + ) -> None: + """Init ``LitestarExtensionGroup``""" + super().__init__(name=name, commands=commands, **attrs) + self._prepare_done = False + + for entry_point in entry_points(group="litestar.commands"): + command = entry_point.load() + _wrap_commands([command]) + self.add_command(command, entry_point.name) + + def _prepare(self, ctx: Context) -> None: + if self._prepare_done: + return + + if isinstance(ctx.obj, LitestarEnv): + env: LitestarEnv | None = ctx.obj + else: + try: + env = ctx.obj = LitestarEnv.from_env(ctx.params.get("app_path"), ctx.params.get("app_dir")) + except LitestarCLIException: + env = None + + if env: + for plugin in env.app.plugins.cli: + plugin.on_cli_init(self) + + self._prepare_done = True + + def make_context( + self, + info_name: str | None, + args: list[str], + parent: Context | None = None, + **extra: Any, + ) -> Context: + ctx = super().make_context(info_name, args, parent, **extra) + self._prepare(ctx) + return ctx + + def list_commands(self, ctx: Context) -> list[str]: + self._prepare(ctx) + return super().list_commands(ctx) + + +def _inject_args(func: Callable[P, T]) -> Callable[P, T]: + """Inject the app instance into a ``Command``""" + params = inspect.signature(func).parameters + + @wraps(func) + def wrapped(ctx: Context, /, *args: P.args, **kwargs: P.kwargs) -> T: + needs_app = "app" in params + needs_env = "env" in params + if needs_env or needs_app: + # only resolve this if actually requested. Commands that don't need an env or app should be able to run + # without + if not isinstance(ctx.obj, LitestarEnv): + ctx.obj = ctx.obj() + env = ctx.ensure_object(LitestarEnv) + if needs_app: + kwargs["app"] = env.app + if needs_env: + kwargs["env"] = env + + if "ctx" in params: + kwargs["ctx"] = ctx + + return func(*args, **kwargs) + + return pass_context(wrapped) + + +def _wrap_commands(commands: Iterable[Command]) -> None: + for command in commands: + if isinstance(command, Group): + _wrap_commands(command.commands.values()) + elif command.callback: + command.callback = _inject_args(command.callback) + + +def _bool_from_env(key: str, default: bool = False) -> bool: + value = getenv(key) + if not value: + return default + value = value.lower() + return value in ("true", "1") + + +def _load_app_from_path(app_path: str) -> LoadedApp: + module_path, app_name = app_path.split(":") + module = importlib.import_module(module_path) + app = getattr(module, app_name) + is_factory = False + if not isinstance(app, Litestar) and callable(app): + app = app() + is_factory = True + return LoadedApp(app=app, app_path=app_path, is_factory=is_factory) + + +def _path_to_dotted_path(path: Path) -> str: + if path.stem == "__init__": + path = path.parent + return ".".join(path.with_suffix("").parts) + + +def _arbitrary_autodiscovery_paths(base_dir: Path) -> Generator[Path, None, None]: + yield from _autodiscovery_paths(base_dir, arbitrary=False) + for path in base_dir.iterdir(): + if path.name.startswith(".") or path.name.startswith("_"): + continue + if path.is_file() and path.suffix == ".py": + yield path + + +def _autodiscovery_paths(base_dir: Path, arbitrary: bool = True) -> Generator[Path, None, None]: + for name in AUTODISCOVERY_FILE_NAMES: + path = base_dir / name + + if path.exists() or path.with_suffix(".py").exists(): + yield path + if arbitrary and path.is_dir(): + yield from _arbitrary_autodiscovery_paths(path) + + +def _autodiscover_app(cwd: Path) -> LoadedApp: + for file_path in _autodiscovery_paths(cwd): + import_path = _path_to_dotted_path(file_path.relative_to(cwd)) + module = importlib.import_module(import_path) + + for attr, value in chain( + [("app", getattr(module, "app", None)), ("application", getattr(module, "application", None))], + module.__dict__.items(), + ): + if isinstance(value, Litestar): + app_string = f"{import_path}:{attr}" + os.environ["LITESTAR_APP"] = app_string + console.print(f"Using Litestar app from [bright_blue]{app_string}") + return LoadedApp(app=value, app_path=app_string, is_factory=False) + + if hasattr(module, "create_app"): + app_string = f"{import_path}:create_app" + os.environ["LITESTAR_APP"] = app_string + console.print(f"Using Litestar factory [bright_blue]{app_string}") + return LoadedApp(app=module.create_app(), app_path=app_string, is_factory=True) + + for attr, value in module.__dict__.items(): + if not callable(value): + continue + return_annotation = ( + get_type_hints(value, include_extras=True).get("return") if hasattr(value, "__annotations__") else None + ) + if not return_annotation: + continue + if return_annotation in ("Litestar", Litestar): + app_string = f"{import_path}:{attr}" + os.environ["LITESTAR_APP"] = app_string + console.print(f"Using Litestar factory [bright_blue]{app_string}") + return LoadedApp(app=value(), app_path=f"{app_string}", is_factory=True) + + raise LitestarCLIException("Could not find a Litestar app or factory") + + +def _format_is_enabled(value: Any) -> str: + """Return a coloured string `"Enabled" if ``value`` is truthy, else "Disabled".""" + return "[green]Enabled[/]" if value else "[red]Disabled[/]" + + +def show_app_info(app: Litestar) -> None: # pragma: no cover + """Display basic information about the application and its configuration.""" + + table = Table(show_header=False) + table.add_column("title", style="cyan") + table.add_column("value", style="bright_blue") + + table.add_row("Litestar version", f"{__version__.major}.{__version__.minor}.{__version__.patch}") + table.add_row("Debug mode", _format_is_enabled(app.debug)) + table.add_row("Python Debugger on exception", _format_is_enabled(app.pdb_on_exception)) + table.add_row("CORS", _format_is_enabled(app.cors_config)) + table.add_row("CSRF", _format_is_enabled(app.csrf_config)) + if app.allowed_hosts: + allowed_hosts = app.allowed_hosts + + table.add_row("Allowed hosts", ", ".join(allowed_hosts.allowed_hosts)) + + openapi_enabled = _format_is_enabled(app.openapi_config) + if app.openapi_config: + openapi_enabled += f" path=[yellow]{app.openapi_config.openapi_controller.path}" + table.add_row("OpenAPI", openapi_enabled) + + table.add_row("Compression", app.compression_config.backend if app.compression_config else "[red]Disabled") + + if app.template_engine: + table.add_row("Template engine", type(app.template_engine).__name__) + + if app.static_files_config: + static_files_configs = app.static_files_config + static_files_info = [ + f"path=[yellow]{static_files.path}[/] dirs=[yellow]{', '.join(map(str, static_files.directories))}[/] " + f"html_mode={_format_is_enabled(static_files.html_mode)}" + for static_files in static_files_configs + ] + table.add_row("Static files", "\n".join(static_files_info)) + + middlewares = [] + for middleware in app.middleware: + updated_middleware = middleware.middleware if isinstance(middleware, DefineMiddleware) else middleware + middlewares.append(get_name(updated_middleware)) + if middlewares: + table.add_row("Middlewares", ", ".join(middlewares)) + + console.print(table) + + +def validate_ssl_file_paths(certfile_arg: str | None, keyfile_arg: str | None) -> tuple[str, str] | tuple[None, None]: + """Validate whether given paths exist, are not directories and were both provided or none was. Return the resolved paths. + + Args: + certfile_arg: path argument for the certificate file + keyfile_arg: path argument for the key file + + Returns: + tuple of resolved paths converted to str or tuple of None's if no argument was provided + """ + if certfile_arg is None and keyfile_arg is None: + return (None, None) + + resolved_paths = [] + + for argname, arg in {"--ssl-certfile": certfile_arg, "--ssl-keyfile": keyfile_arg}.items(): + if arg is None: + raise LitestarCLIException(f"No value provided for {argname}") + path = Path(arg).resolve() + if path.is_dir(): + raise LitestarCLIException(f"Path provided for {argname} is a directory: {path}") + if not path.exists(): + raise LitestarCLIException(f"File provided for {argname} was not found: {path}") + resolved_paths.append(str(path)) + + return tuple(resolved_paths) # type: ignore[return-value] + + +def create_ssl_files( + certfile_arg: str | None, keyfile_arg: str | None, common_name: str = "localhost" +) -> tuple[str, str]: + """Validate whether both files were provided, are not directories, their parent dirs exist and either both files exists or none does. + If neither file exists, create a self-signed ssl certificate and a passwordless key at the location. + + Args: + certfile_arg: path argument for the certificate file + keyfile_arg: path argument for the key file + common_name: the CN to be used as cert issuer and subject + + Returns: + resolved paths of the found or generated files + """ + resolved_paths = [] + + for argname, arg in {"--ssl-certfile": certfile_arg, "--ssl-keyfile": keyfile_arg}.items(): + if arg is None: + raise LitestarCLIException(f"No value provided for {argname}") + path = Path(arg).resolve() + if path.is_dir(): + raise LitestarCLIException(f"Path provided for {argname} is a directory: {path}") + if not (parent_dir := path.parent).exists(): + raise LitestarCLIException( + f"Could not create file, parent directory for {argname} doesn't exist: {parent_dir}" + ) + resolved_paths.append(path) + + if (not resolved_paths[0].exists()) ^ (not resolved_paths[1].exists()): + raise LitestarCLIException( + "Both certificate and key file must exists or both must not exists when using --create-self-signed-cert" + ) + + if (not resolved_paths[0].exists()) and (not resolved_paths[1].exists()): + _generate_self_signed_cert(resolved_paths[0], resolved_paths[1], common_name) + + return (str(resolved_paths[0]), str(resolved_paths[1])) + + +def _generate_self_signed_cert(certfile_path: Path, keyfile_path: Path, common_name: str) -> None: + """Create a self-signed certificate using the cryptography modules at given paths""" + try: + from cryptography import x509 + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + except ImportError as err: + raise LitestarCLIException( + "Cryptography must be installed when using --create-self-signed-cert\nPlease install the litestar[cryptography] extras" + ) from err + + subject = x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, common_name), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Development Certificate"), + ] + ) + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend()) + + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(tz=timezone.utc)) + .not_valid_after(datetime.now(tz=timezone.utc) + timedelta(days=365)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName(common_name)]), critical=False) + .add_extension(x509.ExtendedKeyUsage([x509.OID_SERVER_AUTH]), critical=False) + .sign(key, hashes.SHA256(), default_backend()) + ) + + with certfile_path.open("wb") as cert_file: + cert_file.write(cert.public_bytes(serialization.Encoding.PEM)) + + with keyfile_path.open("wb") as key_file: + key_file.write( + key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + + +def remove_routes_with_patterns( + routes: list[HTTPRoute | ASGIRoute | WebSocketRoute], patterns: tuple[str, ...] +) -> list[HTTPRoute | ASGIRoute | WebSocketRoute]: + regex_routes = [] + valid_patterns = [] + for pattern in patterns: + try: + check_pattern = re.compile(pattern) + valid_patterns.append(check_pattern) + except re.error as e: + console.print(f"Error: {e}. Invalid regex pattern supplied: '{pattern}'. Omitting from querying results.") + + for route in routes: + checked_pattern_route_matches = [] + for pattern_compile in valid_patterns: + matches = pattern_compile.match(route.path) + checked_pattern_route_matches.append(matches) + + if not any(checked_pattern_route_matches): + regex_routes.append(route) + + return regex_routes + + +def remove_default_schema_routes( + routes: list[HTTPRoute | ASGIRoute | WebSocketRoute], openapi_config: OpenAPIConfig +) -> list[HTTPRoute | ASGIRoute | WebSocketRoute]: + schema_path = openapi_config.openapi_controller.path + return remove_routes_with_patterns(routes, (schema_path,)) diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/__init__.py b/venv/lib/python3.11/site-packages/litestar/cli/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..af651ad Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/__init__.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/core.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/core.cpython-311.pyc new file mode 100644 index 0000000..6f44d52 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/core.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/schema.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/schema.cpython-311.pyc new file mode 100644 index 0000000..d6d7532 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/schema.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/sessions.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/sessions.cpython-311.pyc new file mode 100644 index 0000000..bebfa1a Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/sessions.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/core.py b/venv/lib/python3.11/site-packages/litestar/cli/commands/core.py new file mode 100644 index 0000000..5b55253 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/commands/core.py @@ -0,0 +1,335 @@ +from __future__ import annotations + +import inspect +import multiprocessing +import os +import subprocess +import sys +from contextlib import AbstractContextManager, ExitStack, contextmanager +from typing import TYPE_CHECKING, Any, Iterator + +import click +from click import Context, command, option +from rich.tree import Tree + +from litestar.app import DEFAULT_OPENAPI_CONFIG +from litestar.cli._utils import ( + UVICORN_INSTALLED, + LitestarEnv, + console, + create_ssl_files, + remove_default_schema_routes, + remove_routes_with_patterns, + show_app_info, + validate_ssl_file_paths, +) +from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute +from litestar.utils.helpers import unwrap_partial + +__all__ = ("info_command", "routes_command", "run_command") + +if TYPE_CHECKING: + from litestar import Litestar + + +@contextmanager +def _server_lifespan(app: Litestar) -> Iterator[None]: + """Context manager handling the ASGI server lifespan. + + It will be entered just before the ASGI server is started through the CLI. + """ + with ExitStack() as exit_stack: + for manager in app._server_lifespan_managers: + if not isinstance(manager, AbstractContextManager): + manager = manager(app) # type: ignore[assignment] + exit_stack.enter_context(manager) # type: ignore[arg-type] + + yield + + +def _convert_uvicorn_args(args: dict[str, Any]) -> list[str]: + process_args = [] + for arg, value in args.items(): + if isinstance(value, bool): + if value: + process_args.append(f"--{arg}") + elif isinstance(value, tuple): + process_args.extend(f"--{arg}={item}" for item in value) + else: + process_args.append(f"--{arg}={value}") + + return process_args + + +def _run_uvicorn_in_subprocess( + *, + env: LitestarEnv, + host: str | None, + port: int | None, + workers: int | None, + reload: bool, + reload_dirs: tuple[str, ...] | None, + reload_include: tuple[str, ...] | None, + reload_exclude: tuple[str, ...] | None, + fd: int | None, + uds: str | None, + certfile_path: str | None, + keyfile_path: str | None, +) -> None: + process_args: dict[str, Any] = { + "reload": reload, + "host": host, + "port": port, + "workers": workers, + "factory": env.is_app_factory, + } + if fd is not None: + process_args["fd"] = fd + if uds is not None: + process_args["uds"] = uds + if reload_dirs: + process_args["reload-dir"] = reload_dirs + if reload_include: + process_args["reload-include"] = reload_include + if reload_exclude: + process_args["reload-exclude"] = reload_exclude + if certfile_path is not None: + process_args["ssl-certfile"] = certfile_path + if keyfile_path is not None: + process_args["ssl-keyfile"] = keyfile_path + subprocess.run( + [sys.executable, "-m", "uvicorn", env.app_path, *_convert_uvicorn_args(process_args)], # noqa: S603 + check=True, + ) + + +@command(name="version") +@option("-s", "--short", help="Exclude release level and serial information", is_flag=True, default=False) +def version_command(short: bool) -> None: + """Show the currently installed Litestar version.""" + from litestar import __version__ + + click.echo(__version__.formatted(short=short)) + + +@command(name="info") +def info_command(app: Litestar) -> None: + """Show information about the detected Litestar app.""" + + show_app_info(app) + + +@command(name="run") +@option("-r", "--reload", help="Reload server on changes", default=False, is_flag=True) +@option("-R", "--reload-dir", help="Directories to watch for file changes", multiple=True) +@option( + "-I", "--reload-include", help="Glob patterns for files to include when watching for file changes", multiple=True +) +@option( + "-E", "--reload-exclude", help="Glob patterns for files to exclude when watching for file changes", multiple=True +) +@option("-p", "--port", help="Serve under this port", type=int, default=8000, show_default=True) +@option( + "-W", + "--wc", + "--web-concurrency", + help="The number of HTTP workers to launch", + type=click.IntRange(min=1, max=multiprocessing.cpu_count() + 1), + show_default=True, + default=1, +) +@option("-H", "--host", help="Server under this host", default="127.0.0.1", show_default=True) +@option( + "-F", + "--fd", + "--file-descriptor", + help="Bind to a socket from this file descriptor.", + type=int, + default=None, + show_default=True, +) +@option("-U", "--uds", "--unix-domain-socket", help="Bind to a UNIX domain socket.", default=None, show_default=True) +@option("-d", "--debug", help="Run app in debug mode", is_flag=True) +@option("-P", "--pdb", "--use-pdb", help="Drop into PDB on an exception", is_flag=True) +@option("--ssl-certfile", help="Location of the SSL cert file", default=None) +@option("--ssl-keyfile", help="Location of the SSL key file", default=None) +@option( + "--create-self-signed-cert", + help="If certificate and key are not found at specified locations, create a self-signed certificate and a key", + is_flag=True, +) +def run_command( + reload: bool, + port: int, + wc: int, + host: str, + fd: int | None, + uds: str | None, + debug: bool, + reload_dir: tuple[str, ...], + reload_include: tuple[str, ...], + reload_exclude: tuple[str, ...], + pdb: bool, + ssl_certfile: str | None, + ssl_keyfile: str | None, + create_self_signed_cert: bool, + ctx: Context, +) -> None: + """Run a Litestar app; requires ``uvicorn``. + + The app can be either passed as a module path in the form of .:, + set as an environment variable LITESTAR_APP with the same format or automatically discovered from one of these + canonical paths: app.py, asgi.py, application.py or app/__init__.py. When auto-discovering application factories, + functions with the name ``create_app`` are considered, or functions that are annotated as returning a ``Litestar`` + instance. + """ + + if debug: + os.environ["LITESTAR_DEBUG"] = "1" + + if pdb: + os.environ["LITESTAR_PDB"] = "1" + + if not UVICORN_INSTALLED: + console.print( + r"uvicorn is not installed. Please install the standard group, litestar\[standard], to use this command." + ) + sys.exit(1) + + if callable(ctx.obj): + ctx.obj = ctx.obj() + else: + if debug: + ctx.obj.app.debug = True + if pdb: + ctx.obj.app.pdb_on_exception = True + + env: LitestarEnv = ctx.obj + app = env.app + + reload_dirs = env.reload_dirs or reload_dir + reload_include = env.reload_include or reload_include + reload_exclude = env.reload_exclude or reload_exclude + + host = env.host or host + port = env.port if env.port is not None else port + fd = env.fd if env.fd is not None else fd + uds = env.uds or uds + reload = env.reload or reload or bool(reload_dirs) or bool(reload_include) or bool(reload_exclude) + workers = env.web_concurrency or wc + + ssl_certfile = ssl_certfile or env.certfile_path + ssl_keyfile = ssl_keyfile or env.keyfile_path + create_self_signed_cert = create_self_signed_cert or env.create_self_signed_cert + + certfile_path, keyfile_path = ( + create_ssl_files(ssl_certfile, ssl_keyfile, host) + if create_self_signed_cert + else validate_ssl_file_paths(ssl_certfile, ssl_keyfile) + ) + + console.rule("[yellow]Starting server process", align="left") + + show_app_info(app) + with _server_lifespan(app): + if workers == 1 and not reload: + import uvicorn + + # A guard statement at the beginning of this function prevents uvicorn from being unbound + # See "reportUnboundVariable in: + # https://microsoft.github.io/pyright/#/configuration?id=type-check-diagnostics-settings + uvicorn.run( # pyright: ignore + app=env.app_path, + host=host, + port=port, + fd=fd, + uds=uds, + factory=env.is_app_factory, + ssl_certfile=certfile_path, + ssl_keyfile=keyfile_path, + ) + else: + # invoke uvicorn in a subprocess to be able to use the --reload flag. see + # https://github.com/litestar-org/litestar/issues/1191 and https://github.com/encode/uvicorn/issues/1045 + if sys.gettrace() is not None: + console.print( + "[yellow]Debugger detected. Breakpoints might not work correctly inside route handlers when running" + " with the --reload or --workers options[/]" + ) + + _run_uvicorn_in_subprocess( + env=env, + host=host, + port=port, + workers=workers, + reload=reload, + reload_dirs=reload_dirs, + reload_include=reload_include, + reload_exclude=reload_exclude, + fd=fd, + uds=uds, + certfile_path=certfile_path, + keyfile_path=keyfile_path, + ) + + +@command(name="routes") +@option("--schema", help="Include schema routes", is_flag=True, default=False) +@option("--exclude", help="routes to exclude via regex", type=str, is_flag=False, multiple=True) +def routes_command(app: Litestar, exclude: tuple[str, ...], schema: bool) -> None: # pragma: no cover + """Display information about the application's routes.""" + + sorted_routes = sorted(app.routes, key=lambda r: r.path) + if not schema: + openapi_config = app.openapi_config or DEFAULT_OPENAPI_CONFIG + sorted_routes = remove_default_schema_routes(sorted_routes, openapi_config) + if exclude is not None: + sorted_routes = remove_routes_with_patterns(sorted_routes, exclude) + + console.print(_RouteTree(sorted_routes)) + + +class _RouteTree(Tree): + def __init__(self, routes: list[HTTPRoute | ASGIRoute | WebSocketRoute]) -> None: + super().__init__("", hide_root=True) + self._routes = routes + self._build() + + def _build(self) -> None: + for route in self._routes: + if isinstance(route, HTTPRoute): + self._handle_http_route(route) + elif isinstance(route, WebSocketRoute): + self._handle_websocket_route(route) + else: + self._handle_asgi_route(route) + + def _handle_asgi_like_route(self, route: ASGIRoute | WebSocketRoute, route_type: str) -> None: + branch = self.add(f"[green]{route.path}[/green] ({route_type})") + branch.add(f"[blue]{route.route_handler.name or route.route_handler.handler_name}[/blue]") + + def _handle_asgi_route(self, route: ASGIRoute) -> None: + self._handle_asgi_like_route(route, route_type="ASGI") + + def _handle_websocket_route(self, route: WebSocketRoute) -> None: + self._handle_asgi_like_route(route, route_type="WS") + + def _handle_http_route(self, route: HTTPRoute) -> None: + branch = self.add(f"[green]{route.path}[/green] (HTTP)") + for handler in route.route_handlers: + handler_info = [ + f"[blue]{handler.name or handler.handler_name}[/blue]", + ] + + if inspect.iscoroutinefunction(unwrap_partial(handler.fn)): + handler_info.append("[magenta]async[/magenta]") + else: + handler_info.append("[yellow]sync[/yellow]") + + handler_info.append(f'[cyan]{", ".join(sorted(handler.http_methods))}[/cyan]') + + if len(handler.paths) > 1: + for path in handler.paths: + branch.add(" ".join([f"[green]{path}[green]", *handler_info])) + else: + branch.add(" ".join(handler_info)) diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/schema.py b/venv/lib/python3.11/site-packages/litestar/cli/commands/schema.py new file mode 100644 index 0000000..a323bc7 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/commands/schema.py @@ -0,0 +1,82 @@ +from pathlib import Path + +import msgspec +from click import Path as ClickPath +from click import group, option +from yaml import dump as dump_yaml + +from litestar import Litestar +from litestar._openapi.typescript_converter.converter import ( + convert_openapi_to_typescript, +) +from litestar.cli._utils import JSBEAUTIFIER_INSTALLED, LitestarCLIException, LitestarGroup +from litestar.serialization import encode_json, get_serializer + +__all__ = ("generate_openapi_schema", "generate_typescript_specs", "schema_group") + + +@group(cls=LitestarGroup, name="schema") +def schema_group() -> None: + """Manage server-side OpenAPI schemas.""" + + +def _generate_openapi_schema(app: Litestar, output: Path) -> None: + """Generate an OpenAPI Schema.""" + serializer = get_serializer(app.type_encoders) + if output.suffix in (".yml", ".yaml"): + content = dump_yaml( + msgspec.to_builtins(app.openapi_schema.to_schema(), enc_hook=serializer), + default_flow_style=False, + encoding="utf-8", + ) + else: + content = msgspec.json.format( + encode_json(app.openapi_schema.to_schema(), serializer=serializer), + indent=4, + ) + + try: + output.write_bytes(content) + except OSError as e: # pragma: no cover + raise LitestarCLIException(f"failed to write schema to path {output}") from e + + +@schema_group.command("openapi") # type: ignore[misc] +@option( + "--output", + help="output file path", + type=ClickPath(dir_okay=False, path_type=Path), + default=Path("openapi_schema.json"), + show_default=True, +) +def generate_openapi_schema(app: Litestar, output: Path) -> None: + """Generate an OpenAPI Schema.""" + _generate_openapi_schema(app, output) + + +@schema_group.command("typescript") # type: ignore[misc] +@option( + "--output", + help="output file path", + type=ClickPath(dir_okay=False, path_type=Path), + default=Path("api-specs.ts"), + show_default=True, +) +@option("--namespace", help="namespace to use for the typescript specs", type=str, default="API") +def generate_typescript_specs(app: Litestar, output: Path, namespace: str) -> None: + """Generate TypeScript specs from the OpenAPI schema.""" + if JSBEAUTIFIER_INSTALLED: # pragma: no cover + from jsbeautifier import Beautifier + + beautifier = Beautifier() + else: + beautifier = None + try: + specs = convert_openapi_to_typescript(app.openapi_schema, namespace) + # beautifier will be defined if JSBEAUTIFIER_INSTALLED is True + specs_output = ( + beautifier.beautify(specs.write()) if JSBEAUTIFIER_INSTALLED and beautifier else specs.write() # pyright: ignore + ) + output.write_text(specs_output) + except OSError as e: # pragma: no cover + raise LitestarCLIException(f"failed to write schema to path {output}") from e diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py b/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py new file mode 100644 index 0000000..f048dd1 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py @@ -0,0 +1,58 @@ +from click import argument, group +from rich.prompt import Confirm + +from litestar import Litestar +from litestar.cli._utils import LitestarCLIException, LitestarGroup, console +from litestar.middleware import DefineMiddleware +from litestar.middleware.session import SessionMiddleware +from litestar.middleware.session.server_side import ServerSideSessionBackend +from litestar.utils import is_class_and_subclass + +__all__ = ("clear_sessions_command", "delete_session_command", "get_session_backend", "sessions_group") + + +def get_session_backend(app: Litestar) -> ServerSideSessionBackend: + """Get the session backend used by a ``Litestar`` app.""" + for middleware in app.middleware: + if isinstance(middleware, DefineMiddleware): + if not is_class_and_subclass(middleware.middleware, SessionMiddleware): + continue + backend = middleware.kwargs["backend"] + if not isinstance(backend, ServerSideSessionBackend): + raise LitestarCLIException("Only server-side backends are supported") + return backend + raise LitestarCLIException("Session middleware not installed") + + +@group(cls=LitestarGroup, name="sessions") +def sessions_group() -> None: + """Manage server-side sessions.""" + + +@sessions_group.command("delete") # type: ignore[misc] +@argument("session-id") +def delete_session_command(session_id: str, app: Litestar) -> None: + """Delete a specific session.""" + import anyio + + backend = get_session_backend(app) + store = backend.config.get_store_from_app(app) + + if Confirm.ask(f"Delete session {session_id!r}?"): + anyio.run(backend.delete, session_id, store) + console.print(f"[green]Deleted session {session_id!r}") + + +@sessions_group.command("clear") # type: ignore[misc] +def clear_sessions_command(app: Litestar) -> None: + """Delete all sessions.""" + import anyio + + backend = get_session_backend(app) + store = backend.config.get_store_from_app(app) + if not hasattr(store, "delete_all"): + raise LitestarCLIException(f"{type(store)} does not support clearing all sessions") + + if Confirm.ask("[red]Delete all sessions?"): + anyio.run(store.delete_all) # pyright: ignore + console.print("[green]All active sessions deleted") diff --git a/venv/lib/python3.11/site-packages/litestar/cli/main.py b/venv/lib/python3.11/site-packages/litestar/cli/main.py new file mode 100644 index 0000000..32505f6 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/cli/main.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from pathlib import Path + +from click import Context, group, option, pass_context +from click import Path as ClickPath + +from ._utils import LitestarEnv, LitestarExtensionGroup +from .commands import core, schema, sessions + +__all__ = ("litestar_group",) + + +@group(cls=LitestarExtensionGroup, context_settings={"help_option_names": ["-h", "--help"]}) +@option("--app", "app_path", help="Module path to a Litestar application") +@option( + "--app-dir", + help="Look for APP in the specified directory, by adding this to the PYTHONPATH. Defaults to the current working directory.", + default=None, + type=ClickPath(dir_okay=True, file_okay=False, path_type=Path), + show_default=False, +) +@pass_context +def litestar_group(ctx: Context, app_path: str | None, app_dir: Path | None = None) -> None: + """Litestar CLI.""" + if ctx.obj is None: # env has not been loaded yet, so we can lazy load it + ctx.obj = lambda: LitestarEnv.from_env(app_path, app_dir=app_dir) + + +# add sub commands here + +litestar_group.add_command(core.info_command) +litestar_group.add_command(core.run_command) +litestar_group.add_command(core.routes_command) +litestar_group.add_command(core.version_command) +litestar_group.add_command(sessions.sessions_group) +litestar_group.add_command(schema.schema_group) -- cgit v1.2.3