diff --git a/litestar/app.py b/litestar/app.py index fa49afdb39..0371924a97 100644 --- a/litestar/app.py +++ b/litestar/app.py @@ -4,7 +4,6 @@ import inspect import itertools import logging -import os import pdb # noqa: T100 import warnings from collections import defaultdict @@ -53,7 +52,7 @@ from litestar.stores.registry import StoreRegistry from litestar.types import Empty, TypeDecodersSequence from litestar.types.internal_types import PathParameterDefinition, RouteHandlerMapItem, TemplateConfigType -from litestar.utils import ensure_async_callable, join_paths, unique +from litestar.utils import ensure_async_callable, envflag, join_paths, unique from litestar.utils.dataclass import extract_dataclass_items from litestar.utils.predicates import is_async_callable, is_class_and_subclass from litestar.utils.warnings import warn_pdb_on_exception @@ -334,10 +333,10 @@ def __init__( logging_config = LoggingConfig() if debug is None: - debug = os.getenv("LITESTAR_DEBUG", "0") == "1" + debug = envflag("LITESTAR_DEBUG") if pdb_on_exception is None: - pdb_on_exception = os.getenv("LITESTAR_PDB", "0") == "1" + pdb_on_exception = envflag("LITESTAR_PDB") config = AppConfig( after_exception=list(after_exception or []), diff --git a/litestar/cli/_utils.py b/litestar/cli/_utils.py index 729ecabb94..eb39b3fa9e 100644 --- a/litestar/cli/_utils.py +++ b/litestar/cli/_utils.py @@ -30,7 +30,7 @@ from litestar import Litestar, __version__ from litestar.middleware import DefineMiddleware -from litestar.utils import get_name +from litestar.utils import envflag, get_name if sys.version_info >= (3, 10): from importlib.metadata import entry_points @@ -113,7 +113,7 @@ def from_env(cls, app_path: str | None, app_dir: Path | None = None) -> Litestar dotenv.load_dotenv() app_path = app_path or getenv("LITESTAR_APP") app_name = getenv("LITESTAR_APP_NAME") or "Litestar" - quiet_console = getenv("LITESTAR_QUIET_CONSOLE") or False + quiet_console = envflag("LITESTAR_QUIET_CONSOLE") if app_path and getenv("LITESTAR_APP") is None: os.environ["LITESTAR_APP"] = app_path if app_path: @@ -345,7 +345,7 @@ def _autodiscovery_paths(base_dir: Path, arbitrary: bool = True) -> Generator[Pa def _autodiscover_app(cwd: Path) -> LoadedApp: app_name = getenv("LITESTAR_APP_NAME") or "Litestar" - quiet_console = getenv("LITESTAR_QUIET_CONSOLE") or False + quiet_console = envflag("LITESTAR_QUIET_CONSOLE") for file_path in _autodiscovery_paths(cwd): import_path = _path_to_dotted_path(file_path.relative_to(cwd)) module = importlib.import_module(import_path) diff --git a/litestar/cli/commands/core.py b/litestar/cli/commands/core.py index b0a9753325..30beea0b0a 100644 --- a/litestar/cli/commands/core.py +++ b/litestar/cli/commands/core.py @@ -8,6 +8,8 @@ from contextlib import AbstractContextManager, ExitStack, contextmanager from typing import TYPE_CHECKING, Any +from litestar.utils.helpers import envflag + try: import rich_click as click except ImportError: @@ -119,7 +121,14 @@ class CommaSplittedPath(click.Path): @click.command(name="version") -@click.option("-s", "--short", help="Exclude release level and serial information", is_flag=True, default=False) +@click.option( + "-s", + "--short", + help="Exclude release level and serial information", + type=click.BOOL, + default=False, + is_flag=True, +) def version_command(short: bool) -> None: """Show the currently installed Litestar version.""" from litestar import __version__ @@ -135,7 +144,15 @@ def info_command(app: Litestar) -> None: @click.command(name="run") -@click.option("-r", "--reload", help="Reload server on changes", default=False, is_flag=True, envvar="LITESTAR_RELOAD") +@click.option( + "-r", + "--reload", + help="Reload server on changes", + envvar="LITESTAR_RELOAD", + type=click.BOOL, + default=False, + is_flag=True, +) @click.option( "-R", "--reload-dir", @@ -195,15 +212,39 @@ def info_command(app: Litestar) -> None: show_default=True, envvar="LITESTAR_UNIX_DOMAIN_SOCKET", ) -@click.option("-d", "--debug", help="Run app in debug mode", is_flag=True, envvar="LITESTAR_DEBUG") -@click.option("-P", "--pdb", "--use-pdb", help="Drop into PDB on an exception", is_flag=True, envvar="LITESTAR_PDB") +@click.option( + "-d", + "--debug", + help="Run app in debug mode", + envvar="LITESTAR_DEBUG", + type=click.BOOL, + is_flag=True, +) +@click.option( + "-P", + "--pdb", + "--use-pdb", + help="Drop into PDB on an exception", + envvar="LITESTAR_PDB", + type=click.BOOL, + is_flag=True, +) @click.option("--ssl-certfile", help="Location of the SSL cert file", default=None, envvar="LITESTAR_SSL_CERT_PATH") @click.option("--ssl-keyfile", help="Location of the SSL key file", default=None, envvar="LITESTAR_SSL_KEY_PATH") @click.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, envvar="LITESTAR_CREATE_SELF_SIGNED_CERT", + type=click.BOOL, + is_flag=True, +) +@click.option( + "-q", + "--quiet-console", + envvar="LITESTAR_QUIET_CONSOLE", + help="Suppress formatted console output (useful for non-TTY environments, logs, and CI/CD", + type=click.BOOL, + is_flag=True, ) def run_command( reload: bool, @@ -220,6 +261,7 @@ def run_command( ssl_certfile: str | None, ssl_keyfile: str | None, create_self_signed_cert: bool, + quiet_console: bool, ctx: click.Context, ) -> None: """Run a Litestar app. (requires 'uvicorn' to be installed). @@ -234,7 +276,9 @@ def run_command( if pdb: os.environ["LITESTAR_PDB"] = "1" - quiet_console = os.getenv("LITESTAR_QUIET_CONSOLE") or False + + quiet_console = bool(envflag("LITESTAR_QUIET_CONSOLE")) + if not UVICORN_INSTALLED: console.print( r"uvicorn is not installed. Please install the standard group, litestar\[standard], to use this command." @@ -307,7 +351,7 @@ def run_command( @click.command(name="routes") -@click.option("--schema", help="Include schema routes", is_flag=True, default=False) +@click.option("--schema", help="Include schema routes", is_flag=True, default=False, type=click.BOOL) @click.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.""" diff --git a/litestar/utils/__init__.py b/litestar/utils/__init__.py index 4cc9d1429b..3fccb3dd3f 100644 --- a/litestar/utils/__init__.py +++ b/litestar/utils/__init__.py @@ -1,6 +1,6 @@ from litestar.utils.deprecation import deprecated, warn_deprecation -from .helpers import get_enum_string_value, get_name, unique_name_for_scope, url_quote +from .helpers import envflag, get_enum_string_value, get_name, unique_name_for_scope, url_quote from .path import join_paths, normalize_path from .predicates import ( is_annotated_type, @@ -29,6 +29,7 @@ "AsyncIteratorWrapper", "deprecated", "ensure_async_callable", + "envflag", "find_index", "get_enum_string_value", "get_name", diff --git a/litestar/utils/helpers.py b/litestar/utils/helpers.py index 10602e2a00..ada18a8de3 100644 --- a/litestar/utils/helpers.py +++ b/litestar/utils/helpers.py @@ -1,10 +1,12 @@ from __future__ import annotations +import os from enum import Enum from functools import partial from typing import TYPE_CHECKING, TypeVar, cast from urllib.parse import quote +from litestar.exceptions import LitestarException from litestar.utils.typing import get_origin_or_inner_type if TYPE_CHECKING: @@ -102,3 +104,33 @@ def get_exception_group() -> type[BaseException]: from exceptiongroup import ExceptionGroup as _ExceptionGroup # pyright: ignore return cast("type[BaseException]", _ExceptionGroup) + + +def envflag(varname: str) -> bool | None: + """Parse an environment variable as a boolean flag. + + Args: + varname: The name of the environment variable to check. + + Returns: + True for truthy values (1, true, t, yes, y, on), + False for falsy values (0, false, f, no, n, off), + or empty string, None if not set. + + Raises: + LitestarException: If the value is not a recognized boolean. + """ + if varname not in os.environ: + return None + + envvar = os.environ.get(varname) + if not envvar: + return False + + norm = envvar.strip().lower() + if norm in {"1", "true", "t", "yes", "y", "on"}: + return True + if norm in {"0", "false", "f", "no", "n", "off"}: + return False + + raise LitestarException(f"Invalid value for {varname}: '{norm}' is not a valid boolean.") diff --git a/litestar/utils/warnings.py b/litestar/utils/warnings.py index 2322245ce6..1337a4ac30 100644 --- a/litestar/utils/warnings.py +++ b/litestar/utils/warnings.py @@ -1,10 +1,10 @@ from __future__ import annotations -import os import warnings from typing import TYPE_CHECKING from litestar.exceptions import LitestarWarning +from litestar.utils import envflag if TYPE_CHECKING: import re @@ -13,7 +13,7 @@ def warn_implicit_sync_to_thread(source: AnyCallable, stacklevel: int = 2) -> None: - if os.getenv("LITESTAR_WARN_IMPLICIT_SYNC_TO_THREAD") == "0": + if not envflag("LITESTAR_WARN_IMPLICIT_SYNC_TO_THREAD"): return warnings.warn( @@ -29,7 +29,7 @@ def warn_implicit_sync_to_thread(source: AnyCallable, stacklevel: int = 2) -> No def warn_sync_to_thread_with_async_callable(source: AnyCallable, stacklevel: int = 2) -> None: - if os.getenv("LITESTAR_WARN_SYNC_TO_THREAD_WITH_ASYNC") == "0": + if not envflag("LITESTAR_WARN_SYNC_TO_THREAD_WITH_ASYNC"): return warnings.warn( @@ -42,7 +42,7 @@ def warn_sync_to_thread_with_async_callable(source: AnyCallable, stacklevel: int def warn_sync_to_thread_with_generator(source: AnyGenerator, stacklevel: int = 2) -> None: - if os.getenv("LITESTAR_WARN_SYNC_TO_THREAD_WITH_GENERATOR") == "0": + if not envflag("LITESTAR_WARN_SYNC_TO_THREAD_WITH_GENERATOR"): return warnings.warn( @@ -73,7 +73,7 @@ def warn_middleware_excluded_on_all_routes( def warn_signature_namespace_override(signature_key: str, stacklevel: int = 2) -> None: - if os.getenv("LITESTAR_WARN_SIGNATURE_NAMESPACE_OVERRIDE") == "0": + if not envflag("LITESTAR_WARN_SIGNATURE_NAMESPACE_OVERRIDE"): return warnings.warn( diff --git a/tests/unit/test_utils/test_helpers.py b/tests/unit/test_utils/test_helpers.py index 988b4ac512..aaefcbd7ce 100644 --- a/tests/unit/test_utils/test_helpers.py +++ b/tests/unit/test_utils/test_helpers.py @@ -3,7 +3,8 @@ import pytest -from litestar.utils.helpers import get_name, unique_name_for_scope, unwrap_partial +from litestar.exceptions import LitestarException +from litestar.utils.helpers import envflag, get_name, unique_name_for_scope, unwrap_partial T = TypeVar("T") @@ -45,3 +46,44 @@ def test_unique_name_for_scope() -> None: assert unique_name_for_scope("a", ["a", "a_0", "b"]) == "a_1" assert unique_name_for_scope("b", ["a", "a_0", "b"]) == "b_0" + + +def test_envflag_truthy_values(monkeypatch: pytest.MonkeyPatch) -> None: + for value in ("1", "true", "t", "yes", "y", "on", "YeS", "oN", "TRUE", "T"): + monkeypatch.setenv("TEST_FLAG", value) + assert envflag("TEST_FLAG") is True + monkeypatch.delenv("TEST_FLAG") + + +def test_envflag_falsy_values(monkeypatch: pytest.MonkeyPatch) -> None: + for value in ("0", "false", "f", "no", "n", "off", "", "OfF", "fAlSe", "NO"): + monkeypatch.setenv("TEST_FLAG", value) + assert envflag("TEST_FLAG") is False + monkeypatch.delenv("TEST_FLAG") + + +def test_envflag_invalid_value(monkeypatch: pytest.MonkeyPatch) -> None: + for value in ("2", "Tru", "Fals", "maybe", "invalid", "O"): + monkeypatch.setenv("TEST_FLAG", value) + with pytest.raises(LitestarException): + envflag("TEST_FLAG") + + +def test_envflag_missing() -> None: + assert envflag("NONEXISTENT_VAR") is None + + +def test_envflag_overrides(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TEST_FLAG", "true") + assert envflag("TEST_FLAG") is True + monkeypatch.delenv("TEST_FLAG") + + monkeypatch.setenv("TEST_FLAG", "0") + assert envflag("TEST_FLAG") is False + monkeypatch.delenv("TEST_FLAG") + + +def test_envflag_empty_string(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TEST_FLAG", "") + assert envflag("TEST_FLAG") is False + monkeypatch.delenv("TEST_FLAG") diff --git a/tools/sphinx_ext/run_examples.py b/tools/sphinx_ext/run_examples.py index 75ff281ba6..0537e3fa4d 100644 --- a/tools/sphinx_ext/run_examples.py +++ b/tools/sphinx_ext/run_examples.py @@ -22,6 +22,7 @@ from sphinx.addnodes import highlightlang from litestar import Litestar +from litestar.utils import envflag if TYPE_CHECKING: from collections.abc import Generator @@ -34,7 +35,7 @@ logger = logging.getLogger("sphinx") -ignore_missing_output = os.getenv("LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT", "") == "1" +ignore_missing_output = envflag("LITESTAR_DOCS_IGNORE_MISSING_EXAMPLE_OUTPUT") class StartupError(RuntimeError):