refactor: wrap environment access

scripts_bazel/BUILD

@@ -53,5 +53,7 @@ py_binary(
     srcs = ["traceability_gate.py"],
     main = "traceability_gate.py",
     visibility = ["//visibility:public"],
-    deps = [],
+    deps = [
+        "//src/helper_lib",
+    ],
 )

scripts_bazel/traceability_gate.py

@@ -29,12 +29,14 @@
 
 import argparse
 import json
-import os
 import sys
 from pathlib import Path
 from typing import Any
 
+from src.helper_lib import Environment
+
 _SUPPORTED_SCHEMA_VERSION = "2"
+env = Environment()
 
 
 def _print_type_summary(
@@ -259,9 +261,9 @@ def main() -> int:
         args.fail_on_broken_test_refs = True
 
     metrics_path = Path(args.metrics_json)
-    workspace_dir = os.environ.get("BUILD_WORKSPACE_DIRECTORY", "").strip()
+    workspace_dir = env.optional_path("BUILD_WORKSPACE_DIRECTORY")
     if not metrics_path.is_absolute() and workspace_dir:
-        metrics_path = Path(workspace_dir) / metrics_path
+        metrics_path = workspace_dir / metrics_path
     if not metrics_path.exists():
         print(f"Error: metrics JSON not found: {metrics_path}", file=sys.stderr)
         return 1

src/docs_cli/cli.py

@@ -30,20 +30,13 @@
 )
 
 from src.extensions.score_mounts._resolver import load_mounts_manifest, resolve_walk_dir
-from src.helper_lib import find_ws_root, get_runfiles_dir
+from src.helper_lib import Environment, find_ws_root, get_runfiles_dir
 
 logger = logging.getLogger(__name__)
 
 
 _MODULE_HASH_FILE = ".module_bazel_hash"
-
-
-def get_env(name: str) -> str:
-    val = os.environ.get(name)
-    logger.debug("Env: %s = %s", name, val)
-    if val is None:
-        raise ValueError(f"Environment variable {name} is not set")
-    return val
+env = Environment()
 
 
 def _merged_external_needs() -> str:
@@ -52,8 +45,8 @@ def _merged_external_needs() -> str:
     Both env vars hold JSON lists of Bazel labels; the extension parses the
     resulting `external_needs_source` define uniformly.
     """
-    data = json.loads(get_env("DATA") or "[]")
-    external = json.loads(os.environ.get("EXTERNAL_NEEDS_FILES", "[]") or "[]")
+    data = env.string_list("DATA")
+    external = env.string_list("EXTERNAL_NEEDS_FILES", "[]")
     return json.dumps(data + external)
 
 
@@ -146,8 +139,8 @@ def add_watch_dir(path: Path) -> None:
 
 def sphinx_arguments(ws_root: Path, package_dir: Path, build_dir: Path) -> list[str]:
     """Resolve package sources and Bazel-provided configuration for every builder."""
-    is_bazel_build = os.environ.get("ACTION") == "build_needs_json"
-    source_directory = get_env("SOURCE_DIRECTORY")
+    is_bazel_build = env.get("ACTION", "") == "build_needs_json"
+    source_directory = env.required_path("SOURCE_DIRECTORY")
     base_arguments = [
         str(package_dir / source_directory),
         str(build_dir),
@@ -159,10 +152,10 @@ def sphinx_arguments(ws_root: Path, package_dir: Path, build_dir: Path) -> list[
         # Merge DATA (:needs_json / :docs_sources) with EXTERNAL_NEEDS_FILES
         # (:needs_json_file) into one define consumed by the Sphinx extensions.
         f"--define=external_needs_source={_merged_external_needs()}",
-        f"--define=testcase_source_dirs={os.environ.get('TEST_SOURCES', '[]')}",
+        f"--define=testcase_source_dirs={env.get('TEST_SOURCES', '[]')}",
         # Path to the Bazel-emitted mounts manifest (empty when no mounts are
         # configured); consumed by the score_mounts extension.
-        f"--define=mounts_manifest={os.environ.get('MOUNTS_MANIFEST', '')}",
+        f"--define=mounts_manifest={env.optional_path('MOUNTS_MANIFEST') or ''}",
     ]
 
     if is_bazel_build:
@@ -175,46 +168,43 @@ def sphinx_arguments(ws_root: Path, package_dir: Path, build_dir: Path) -> list[
         # The sandboxed Needs rule transports options as JSON so spaces, quotes and
         # equals signs survive the environment boundary. Append them last so an
         # action-specific value can override one of the shared defaults above.
-        base_arguments.extend(json.loads(os.environ.get("SPHINX_EXTRA_OPTS", "[]")))
+        base_arguments.extend(env.string_list("SPHINX_EXTRA_OPTS", "[]"))
     else:
         # Interactive builds keep warnings in the workspace so developers can
         # inspect them after a failed build. A Bazel action reports failure
         # through its exit code and must leave its declared output tree free of
         # this diagnostic side file.
         base_arguments.extend(["--warning-file", str(build_dir / "warnings.txt")])
 
-    generated_config = os.environ.get("SPHINX_CONFIG_FILE", "")
-    if generated_config:
+    if config_file := env.optional_path("SPHINX_CONFIG_FILE"):
         # The action receives ctx.file.config.path, which is interpreted from
         # the action's execution-root working directory. Resolve it locally
         # instead of using runfiles lookup; interactive targets receive a
         # runfiles-relative path and need that lookup before Sphinx gets the
         # containing directory.
-        config_file = Path(generated_config)
         if is_bazel_build:
             config_file = config_file.absolute()
         elif not config_file.is_absolute():
             config_file = get_runfiles_dir() / config_file
         base_arguments.extend(["-c", str(config_file.parent)])
 
-    metamodel_yaml = os.environ.get("SCORE_METAMODEL_YAML", "")
-    if metamodel_yaml:
+    if metamodel_yaml := env.optional_path("SCORE_METAMODEL_YAML"):
         # Under ``bazel run``, this environment variable is runfiles-relative
         # and must be resolved through RUNFILES_DIR. A sandboxed Needs action
         # instead expands the metamodel label to an execution-root path in
         # SPHINX_EXTRA_OPTS; applying runfiles lookup there would escape the
         # action's declared inputs.
-        if not is_bazel_build and not os.path.isabs(metamodel_yaml):
-            runfiles_dir = os.environ.get("RUNFILES_DIR", "")
-            metamodel_yaml = str(
-                (Path(runfiles_dir) / metamodel_yaml)
-                if runfiles_dir
-                else (ws_root / metamodel_yaml)
+        if not is_bazel_build and not metamodel_yaml.is_absolute():
+            runfiles_dir = env.optional_path("RUNFILES_DIR")
+            metamodel_yaml = (
+                runfiles_dir / metamodel_yaml
+                if runfiles_dir is not None
+                else ws_root / metamodel_yaml
             )
-        metamodel_yaml = os.path.abspath(metamodel_yaml)
+        metamodel_yaml = metamodel_yaml.absolute()
         base_arguments.append(f"--define=score_metamodel_yaml={metamodel_yaml}")
 
-    if github_repository := os.getenv("GITHUB_REPOSITORY"):
+    if github_repository := env.get("GITHUB_REPOSITORY", ""):
         # GITHUB_REPOSITORY is expected as "owner/repo"; partition("/") splits
         # once into (owner, separator, repo), so we can ignore the separator.
         github_user, _, github_repo = github_repository.partition("/")
@@ -224,18 +214,20 @@ def sphinx_arguments(ws_root: Path, package_dir: Path, build_dir: Path) -> list[
         base_arguments.append("-A=github_version=main")
         # doc_path must be repo-relative so the edit URL does not contain the
         # absolute runner filesystem path (e.g. /home/runner/work/…/docs).
-        relative_doc_path = Path(os.environ.get("PACKAGE_DIR", "")) / source_directory
+        relative_doc_path = (
+            env.optional_path("PACKAGE_DIR") or Path()
+        ) / source_directory
         base_arguments.append(f"-A=doc_path={relative_doc_path}")
 
-    if os.getenv("KNOWN_GOOD_JSON"):
-        base_arguments.append(f"--define=KNOWN_GOOD_JSON={get_env('KNOWN_GOOD_JSON')}")
+    if known_good_json := env.optional_path("KNOWN_GOOD_JSON"):
+        base_arguments.append(f"--define=KNOWN_GOOD_JSON={known_good_json}")
 
     return base_arguments
 
 
 def watch_arguments() -> list[str]:
     """Build autobuild options using the same runfiles resolution as Sphinx."""
-    mounts_manifest = os.environ.get("MOUNTS_MANIFEST", "")
+    mounts_manifest = env.optional_path("MOUNTS_MANIFEST")
     watch_arguments: list[str] = []
     if mounts_manifest:
         # ``MOUNTS_MANIFEST`` is runfiles-relative under ``bazel run`` and
@@ -244,7 +236,7 @@ def watch_arguments() -> list[str]:
         manifest_path = (
             get_runfiles_dir() / mounts_manifest
             if ws_root is not None
-            else Path(mounts_manifest)
+            else mounts_manifest
         )
         for watch_dir in mounted_watch_dirs(
             manifest_path,
@@ -284,17 +276,17 @@ def main(argv: list[str] | None = None) -> int:
         logger.info("Waiting for client to connect on port: " + str(args.debug_port))
         debugpy.wait_for_client()
 
-    action = get_env("ACTION")
+    action = env.get("ACTION")
     is_bazel_build = action == "build_needs_json"
-    ws_root = Path(os.getenv("BUILD_WORKSPACE_DIRECTORY", ""))
+    ws_root = env.optional_path("BUILD_WORKSPACE_DIRECTORY") or Path()
     # Docs source and output are resolved relative to the package where docs()
     # was called; an empty PACKAGE_DIR denotes the workspace root.
-    package_dir = ws_root / os.environ.get("PACKAGE_DIR", "")
+    package_dir = ws_root / (env.optional_path("PACKAGE_DIR") or Path())
     build_dir = package_dir / "_build"
     if is_bazel_build:
         # Bazel owns the action's paths; never use the caller's workspace cache.
         package_dir = Path.cwd()
-        build_dir = Path(get_env("OUTPUT_DIRECTORY")).absolute()
+        build_dir = env.required_path("OUTPUT_DIRECTORY").absolute()
 
     sentinel_files = [
         ws_root / "MODULE.bazel",
@@ -353,7 +345,8 @@ def main(argv: list[str] | None = None) -> int:
 if __name__ == "__main__":
     # Extensions need stable runfiles paths even when Sphinx changes directory.
     for variable in ("RUNFILES_DIR", "JAVA_RUNFILES"):
-        if os.environ.get(variable):
-            os.environ[variable] = str(Path(os.environ[variable]).absolute())
+        value = env.optional_path(variable)
+        if value:
+            os.environ[variable] = str(value.absolute())
 
     sys.exit(main())

src/extensions/score_cross_module_compatibility/__init__.py

@@ -13,7 +13,6 @@
 
 import html
 import json
-import os
 import re
 from dataclasses import asdict, dataclass, replace
 from pathlib import Path
@@ -23,10 +22,11 @@
 from sphinx.util import logging
 from sphinx_needs.need_item import NeedItem
 
-from src.helper_lib import find_ws_root, get_runfiles_dir
+from src.helper_lib import Environment, find_ws_root, get_runfiles_dir
 
 _VERSION_CONDITION = re.compile(r"^\s*version\s*==\s*(\d+)\s*$")
 logger = logging.getLogger(__name__)
+env = Environment()
 
 MANDATORY_ATTRIBUTE = "mandatory-attribute"
 MANDATORY_LINK = "mandatory-link"
@@ -220,9 +220,7 @@ def write(self, outdir: str | Path) -> None:
 
 
 def _manifest_path(app: Sphinx) -> Path | None:
-    raw = getattr(app.config, "mounts_manifest", "") or os.environ.get(
-        "MOUNTS_MANIFEST", ""
-    )
+    raw = getattr(app.config, "mounts_manifest", "") or env.get("MOUNTS_MANIFEST", "")
     if not isinstance(raw, str) or not raw.strip():
         return None
     direct = Path(raw)

src/extensions/score_layout/BUILD

@@ -32,5 +32,8 @@ py_library(
     # python package if we ever want to do that.
     imports = ["."],
     visibility = ["//visibility:public"],
-    deps = [requirement("sphinx")],
+    deps = [
+        requirement("sphinx"),
+        "//src/helper_lib",
+    ],
 )

src/extensions/score_layout/__init__.py

@@ -20,9 +20,10 @@
 import sphinx_options
 from sphinx.application import Sphinx
 
-from src.helper_lib import config_setdefault
+from src.helper_lib import Environment, config_setdefault
 
 logger = logging.getLogger(__name__)
+env = Environment()
 
 # TEMP UNTIL UPSTREAM FIX - BEGIN
 # Bug ref: https://github.com/useblocks/sphinx-needs/issues/1913
@@ -118,9 +119,9 @@ def configure_mounted_source_controls(
         return
 
     source_path = source_path.resolve()
-    workspace_directory = os.environ.get("BUILD_WORKSPACE_DIRECTORY")
+    workspace_directory = env.optional_path("BUILD_WORKSPACE_DIRECTORY")
     if workspace_directory:
-        workspace_root = Path(workspace_directory).resolve()
+        workspace_root = workspace_directory.resolve()
         if (
             source_path.is_relative_to(workspace_root)
             and not {"bazel-bin", "bazel-out"}.intersection(source_path.parts)

src/extensions/score_metamodel/__init__.py

@@ -11,7 +11,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
 import importlib
-import os
 import pkgutil
 from collections.abc import Callable
 from pathlib import Path
@@ -35,9 +34,10 @@
     load_metamodel_data as load_metamodel_data,
     validate_mandatory_regexes as validate_mandatory_regexes,
 )
-from src.helper_lib import config_setdefault
+from src.helper_lib import Environment, config_setdefault
 
 logger = logging.get_logger(__name__)
+env = Environment()
 
 local_check_function = Callable[[Sphinx, NeedItem, CheckLogger], None]
 graph_check_function = Callable[[Sphinx, NeedsView, CheckLogger], None]
@@ -110,8 +110,8 @@ def _run_checks(app: Sphinx) -> None:
 
     logger.debug(f"Running checks for {len(needs_all_needs)} needs")
 
-    ws_root = os.environ.get("BUILD_WORKSPACE_DIRECTORY", None)
-    cwd_or_ws_root = Path(ws_root) if ws_root else Path.cwd()
+    ws_root = env.optional_path("BUILD_WORKSPACE_DIRECTORY")
+    cwd_or_ws_root = ws_root if ws_root else Path.cwd()
     prefix = str(Path(app.srcdir).relative_to(cwd_or_ws_root))
 
     log = CheckLogger(logger, prefix, get_reporter(app))

src/extensions/score_metamodel/log.py

@@ -10,7 +10,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 # *******************************************************************************
-import os
 from typing import Any
 
 from docutils.nodes import Node
@@ -19,9 +18,12 @@
 from sphinx_needs.logging import SphinxLoggerAdapter
 from sphinx_needs.need_item import NeedItem
 
+from src.helper_lib import Environment
+
 Location = str | tuple[str | None, int | None] | Node | None
 NewCheck = tuple[str, Location]
 logger = logging.get_logger(__name__)
+env = Environment()
 
 
 class CheckLogger:
@@ -47,7 +49,9 @@ def get(key: str) -> Any:
             # Note: passing the location as a string allows us to use
             # readable relative paths, passing as a tuple results
             # in absolute paths to ~/.cache/.../bazel-out/..
-            if "RUNFILES_DIR" in os.environ or "RUNFILES_MANIFEST_FILE" in os.environ:
+            if env.optional_path("RUNFILES_DIR") or env.optional_path(
+                "RUNFILES_MANIFEST_FILE"
+            ):
                 matching_file = f"{need['docname']}{need['doctype']}"
             else:
                 matching_file = f"{prefix}/{need['docname']}{need['doctype']}"

src/extensions/score_mounts/__init__.py

@@ -33,7 +33,6 @@
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 
 from sphinx.application import Sphinx
@@ -47,7 +46,9 @@
     resolve_source_files,
     resolve_walk_dir,
 )
-from src.helper_lib import find_ws_root, get_runfiles_dir
+from src.helper_lib import Environment, find_ws_root, get_runfiles_dir
+
+env = Environment()
 
 logger = logging.getLogger(__name__)
 
@@ -62,9 +63,7 @@ def _read_manifest(config: Config):
     it is relative to the exec root (``$(location)``). Resolving the path here
     keeps that context branch out of the pure ``_resolver`` module.
     """
-    raw = getattr(config, "mounts_manifest", None) or os.environ.get(
-        "MOUNTS_MANIFEST", None
-    )
+    raw = getattr(config, "mounts_manifest", None) or env.get("MOUNTS_MANIFEST", "")
     if not raw or not raw.strip() or not isinstance(raw, str):
         return None
 

src/extensions/score_source_code_linker/BUILD

@@ -59,6 +59,7 @@ py_library(
     ],
     imports = ["."],
     visibility = ["//visibility:public"],
+    deps = ["//src/helper_lib"],
 )
 
 score_pytest(