diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/cli')
14 files changed, 0 insertions, 1103 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/cli/__init__.py b/venv/lib/python3.11/site-packages/litestar/cli/__init__.py deleted file mode 100644 index f6c366e..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/__init__.py +++ /dev/null @@ -1,29 +0,0 @@ -"""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 Binary files differdeleted file mode 100644 index 808a2fb..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index 4baba11..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/_utils.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index 7ede592..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/__pycache__/main.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/cli/_utils.py b/venv/lib/python3.11/site-packages/litestar/cli/_utils.py deleted file mode 100644 index f36cd77..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/_utils.py +++ /dev/null @@ -1,562 +0,0 @@ -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 deleted file mode 100644 index e69de29..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/__init__.py +++ /dev/null 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 Binary files differdeleted file mode 100644 index af651ad..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index 6f44d52..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/core.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index d6d7532..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/schema.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index bebfa1a..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/__pycache__/sessions.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index 5b55253..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/core.py +++ /dev/null @@ -1,335 +0,0 @@ -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 <module name>.<submodule>:<app instance or factory>, - 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 deleted file mode 100644 index a323bc7..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/schema.py +++ /dev/null @@ -1,82 +0,0 @@ -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 deleted file mode 100644 index f048dd1..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py +++ /dev/null @@ -1,58 +0,0 @@ -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 deleted file mode 100644 index 32505f6..0000000 --- a/venv/lib/python3.11/site-packages/litestar/cli/main.py +++ /dev/null @@ -1,37 +0,0 @@ -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) |