chore: migrate os env call to new package 1/3

ddtrace/__init__.py

@@ -1,4 +1,3 @@
-import os
 import sys
 
 
@@ -20,14 +19,15 @@
 from ._monkey import patch_all  # noqa: E402
 from .internal.compat import PYTHON_VERSION_INFO  # noqa: E402
 from .internal.utils.deprecations import DDTraceDeprecationWarning  # noqa: E402
+from .settings import _env
 from .settings._config import config
 from .version import get_version  # noqa: E402
 
 
 __version__ = get_version()
 
 # TODO: Deprecate accessing tracer from ddtrace.__init__ module in v4.0
-if os.environ.get("_DD_GLOBAL_TRACER_INIT", "true").lower() in ("1", "true"):
+if _env.getenv("_DD_GLOBAL_TRACER_INIT", "true").lower() in ("1", "true"):
     from ddtrace.trace import tracer  # noqa: F401
 
 __all__ = [

ddtrace/_monkey.py

@@ -1,5 +1,4 @@
 import importlib
-import os
 from types import ModuleType
 from typing import TYPE_CHECKING  # noqa:F401
 from typing import Set
@@ -9,6 +8,7 @@
 
 from ddtrace.internal.compat import Path
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
+from ddtrace.settings import _env
 from ddtrace.settings._config import config
 from ddtrace.vendor.debtcollector import deprecate
 from ddtrace.vendor.packaging.specifiers import SpecifierSet
@@ -17,7 +17,7 @@
 from .internal import telemetry
 from .internal.logger import get_logger
 from .internal.utils import formats
-from .internal.utils.deprecations import DDTraceDeprecationWarning  # noqa: E402
+from .internal.utils.deprecations import DDTraceDeprecationWarning
 
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -351,8 +351,8 @@ def _patch_all(**patch_modules: bool) -> None:
     # The enabled setting can be overridden by environment variables
     for module, _enabled in modules.items():
         env_var = "DD_TRACE_%s_ENABLED" % module.upper()
-        if module not in _NOT_PATCHABLE_VIA_ENVVAR and env_var in os.environ:
-            modules[module] = formats.asbool(os.environ[env_var])
+        if module not in _NOT_PATCHABLE_VIA_ENVVAR and (env_val := _env.getenv(env_var)) is not None:
+            modules[module] = formats.asbool(env_val)
 
         # Enable all dependencies for the module
         if modules[module]:

ddtrace/_trace/apm_filter.py

@@ -1,10 +1,10 @@
-import os
 from typing import List
 from typing import Optional
 
 from ddtrace._trace.processor import TraceProcessor
 from ddtrace._trace.span import Span
 from ddtrace.internal.utils.formats import asbool
+from ddtrace.settings import _env
 
 
 class APMTracingEnabledFilter(TraceProcessor):
@@ -14,7 +14,7 @@ class APMTracingEnabledFilter(TraceProcessor):
 
     def __init__(self) -> None:
         super().__init__()
-        self._apm_tracing_enabled = asbool(os.getenv("DD_APM_TRACING_ENABLED", "true"))
+        self._apm_tracing_enabled = asbool(_env.getenv("DD_APM_TRACING_ENABLED", "true"))
 
     def process_trace(self, trace: List[Span]) -> Optional[List[Span]]:
         if not self._apm_tracing_enabled:

ddtrace/_trace/product.py

@@ -1,6 +1,5 @@
 import enum
 import json
-import os
 import typing as t
 
 from envier import En
@@ -9,6 +8,7 @@
 from ddtrace.internal.utils.deprecations import DDTraceDeprecationWarning
 from ddtrace.internal.utils.formats import asbool
 from ddtrace.internal.utils.formats import parse_tags_str
+from ddtrace.settings import _env
 from ddtrace.settings.http import HttpConfig
 from ddtrace.vendor.debtcollector import deprecate
 
@@ -32,7 +32,7 @@ def post_preload():
     if _config.enabled:
         from ddtrace._monkey import _patch_all
 
-        modules_to_patch = os.getenv("DD_PATCH_MODULES")
+        modules_to_patch = _env.getenv("DD_PATCH_MODULES")
         modules_to_str = parse_tags_str(modules_to_patch)
         modules_to_bool = {k: asbool(v) for k, v in modules_to_str.items()}
         _patch_all(**modules_to_bool)

ddtrace/appsec/_iast/__init__.py

@@ -28,13 +28,13 @@ def wrapped_function(wrapped, instance, args, kwargs):
     return wrapped(*args, **kwargs)
 """
 
-import os
 import sys
 import types
 
 from ddtrace.internal import forksafe
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.module import ModuleWatchdog
+from ddtrace.settings import _env
 from ddtrace.settings.asm import config as asm_config
 
 from ._listener import iast_listen
@@ -186,11 +186,11 @@ def _iast_pytest_activation():
     global _iast_propagation_enabled
     if _iast_propagation_enabled:
         return
-    os.environ["DD_IAST_ENABLED"] = os.environ.get("DD_IAST_ENABLED") or "1"
-    os.environ["DD_IAST_REQUEST_SAMPLING"] = os.environ.get("DD_IAST_REQUEST_SAMPLING") or "100.0"
-    os.environ["_DD_APPSEC_DEDUPLICATION_ENABLED"] = os.environ.get("_DD_APPSEC_DEDUPLICATION_ENABLED") or "false"
-    os.environ["DD_IAST_VULNERABILITIES_PER_REQUEST"] = os.environ.get("DD_IAST_VULNERABILITIES_PER_REQUEST") or "1000"
-    os.environ["DD_IAST_MAX_CONCURRENT_REQUESTS"] = os.environ.get("DD_IAST_MAX_CONCURRENT_REQUESTS") or "1000"
+    _env.environ["DD_IAST_ENABLED"] = _env.getenv("DD_IAST_ENABLED", "1")
+    _env.environ["DD_IAST_REQUEST_SAMPLING"] = _env.getenv("DD_IAST_REQUEST_SAMPLING", "100.0")
+    _env.environ["_DD_APPSEC_DEDUPLICATION_ENABLED"] = _env.getenv("_DD_APPSEC_DEDUPLICATION_ENABLED", "false")
+    _env.environ["DD_IAST_VULNERABILITIES_PER_REQUEST"] = _env.getenv("DD_IAST_VULNERABILITIES_PER_REQUEST", "1000")
+    _env.environ["DD_IAST_MAX_CONCURRENT_REQUESTS"] = _env.getenv("DD_IAST_MAX_CONCURRENT_REQUESTS", "1000")
 
     asm_config._iast_request_sampling = 100.0
     asm_config._deduplication_enabled = False

ddtrace/appsec/_iast/_ast/ast_patching.py

@@ -15,6 +15,7 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.module import origin
 from ddtrace.internal.utils.formats import asbool
+from ddtrace.settings import _env
 from ddtrace.settings.asm import config as asm_config
 
 from .visitor import AstVisitor
@@ -292,7 +293,7 @@ def astpatch_module(module: ModuleType) -> Tuple[str, Optional[ast.Module]]:
         iast_compiling_debug_log(f"Empty file: {module_path}")
         return "", None
 
-    if not asbool(os.environ.get(IAST.ENV_NO_DIR_PATCH, "false")):
+    if not asbool(_env.getenv(IAST.ENV_NO_DIR_PATCH, "false")):
         # Add the dir filter so __ddtrace stuff is not returned by dir(module)
         source_text += _DIR_WRAPPER
 

ddtrace/appsec/_iast/_taint_tracking/_vendor/pybind11/pybind11/setup_helpers.py

@@ -74,6 +74,8 @@
 import distutils.ccompiler
 import distutils.errors
 
+from ddtrace.settings import _env
+
 WIN = sys.platform.startswith("win32") and "mingw" not in sysconfig.get_platform()
 MACOS = sys.platform.startswith("darwin")
 STD_TMPL = "/std:c++{}" if WIN else "-std=c++{}"
@@ -150,8 +152,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             cflags += ["/EHsc", "/bigobj"]
         else:
             cflags += ["-fvisibility=hidden"]
-            env_cflags = os.environ.get("CFLAGS", "")
-            env_cppflags = os.environ.get("CPPFLAGS", "")
+            env_cflags = _env.getenv("CFLAGS", "")
+            env_cppflags = _env.getenv("CPPFLAGS", "")
             c_cpp_flags = shlex.split(env_cflags) + shlex.split(env_cppflags)
             if not any(opt.startswith("-g") for opt in c_cpp_flags):
                 cflags += ["-g0"]
@@ -171,9 +173,7 @@ def cxx_std(self) -> int:
     @cxx_std.setter
     def cxx_std(self, level: int) -> None:
         if self._cxx_level:
-            warnings.warn(
-                "You cannot safely change the cxx_level after setting it!", stacklevel=2
-            )
+            warnings.warn("You cannot safely change the cxx_level after setting it!", stacklevel=2)
 
         # MSVC 2015 Update 3 and later only have 14 (and later 17) modes, so
         # force a valid flag here.
@@ -188,7 +188,7 @@ def cxx_std(self, level: int) -> None:
         cflags = [STD_TMPL.format(level)]
         ldflags = []
 
-        if MACOS and "MACOSX_DEPLOYMENT_TARGET" not in os.environ:
+        if MACOS and _env.getenv("MACOSX_DEPLOYMENT_TARGET") is None:
             # C++17 requires a higher min version of macOS. An earlier version
             # (10.12 or 10.13) can be set manually via environment variable if
             # you are careful in your feature usage, but 10.14 is the safest
@@ -287,9 +287,7 @@ def build_extensions(self) -> None:
         super().build_extensions()
 
 
-def intree_extensions(
-    paths: Iterable[str], package_dir: Optional[Dict[str, str]] = None
-) -> List[Pybind11Extension]:
+def intree_extensions(paths: Iterable[str], package_dir: Optional[Dict[str, str]] = None) -> List[Pybind11Extension]:
     """
     Generate Pybind11Extensions from source files directly located in a Python
     source tree.
@@ -320,10 +318,7 @@ def intree_extensions(
                 exts.append(Pybind11Extension(qualified_name, [path]))
                 break
         else:
-            msg = (
-                f"path {path} is not a child of any of the directories listed "
-                f"in 'package_dir' ({package_dir})"
-            )
+            msg = f"path {path} is not a child of any of the directories listed " f"in 'package_dir' ({package_dir})"
             raise ValueError(msg)
 
     return exts
@@ -447,7 +442,7 @@ def compile_function(
 
             # Determine the number of compilation threads, unless set by an environment variable.
             if self.envvar is not None:
-                threads = int(os.environ.get(self.envvar, self.default))
+                threads = int(_env.getenv(self.envvar, self.default))
 
             def _single_compile(obj: Any) -> None:
                 try:

ddtrace/appsec/_iast/reporter.py

@@ -2,7 +2,6 @@
 from functools import reduce
 import json
 import operator
-import os
 from typing import Any
 from typing import Dict
 from typing import List
@@ -20,6 +19,7 @@
 from ddtrace.appsec._iast.constants import VULN_WEAK_CIPHER_TYPE
 from ddtrace.appsec._iast.constants import VULN_WEAK_RANDOMNESS
 from ddtrace.internal.logger import get_logger
+from ddtrace.settings import _env
 
 
 log = get_logger(__name__)
@@ -99,7 +99,9 @@ class Vulnerability:
     type: str
     evidence: Evidence
     location: Location
-    hash: int = dataclasses.field(init=False, compare=False, hash=("PYTEST_CURRENT_TEST" in os.environ), repr=False)
+    hash: int = dataclasses.field(
+        init=False, compare=False, hash=(_env.getenv("PYTEST_CURRENT_TEST") is not None), repr=False
+    )
 
     def __post_init__(self):
         self.hash = zlib.crc32(repr(self).encode())

ddtrace/appsec/_iast/taint_sinks/weak_cipher.py

@@ -1,4 +1,3 @@
-import os
 from typing import Any
 from typing import Callable
 from typing import Set
@@ -13,6 +12,7 @@
 from ddtrace.appsec._iast.constants import RC4_DEF
 from ddtrace.appsec._iast.constants import VULN_WEAK_CIPHER_TYPE
 from ddtrace.internal.logger import get_logger
+from ddtrace.settings import _env
 from ddtrace.settings.asm import config as asm_config
 
 from .._logs import iast_error
@@ -28,7 +28,7 @@
 
 def get_weak_cipher_algorithms() -> Set:
     CONFIGURED_WEAK_CIPHER_ALGORITHMS = None
-    DD_IAST_WEAK_CIPHER_ALGORITHMS = os.getenv("DD_IAST_WEAK_CIPHER_ALGORITHMS")
+    DD_IAST_WEAK_CIPHER_ALGORITHMS = _env.getenv("DD_IAST_WEAK_CIPHER_ALGORITHMS")
     if DD_IAST_WEAK_CIPHER_ALGORITHMS:
         CONFIGURED_WEAK_CIPHER_ALGORITHMS = set(
             algo.strip() for algo in DD_IAST_WEAK_CIPHER_ALGORITHMS.lower().split(",")

ddtrace/appsec/_iast/taint_sinks/weak_hash.py

@@ -1,9 +1,9 @@
-import os
 from typing import Any
 from typing import Callable
 from typing import Set
 
 from ddtrace.internal.logger import get_logger
+from ddtrace.settings import _env
 from ddtrace.settings.asm import config as asm_config
 
 from ..._common_module_patches import try_unwrap
@@ -28,7 +28,7 @@
 
 def get_weak_hash_algorithms() -> Set:
     CONFIGURED_WEAK_HASH_ALGORITHMS = None
-    DD_IAST_WEAK_HASH_ALGORITHMS = os.getenv("DD_IAST_WEAK_HASH_ALGORITHMS")
+    DD_IAST_WEAK_HASH_ALGORITHMS = _env.getenv("DD_IAST_WEAK_HASH_ALGORITHMS")
     if DD_IAST_WEAK_HASH_ALGORITHMS:
         CONFIGURED_WEAK_HASH_ALGORITHMS = set(algo.strip() for algo in DD_IAST_WEAK_HASH_ALGORITHMS.lower().split(","))
 

ddtrace/appsec/_processor.py

@@ -1,8 +1,6 @@
 import dataclasses
 import errno
 from json.decoder import JSONDecodeError
-import os
-import os.path
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import ClassVar
@@ -43,6 +41,7 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.rate_limiter import RateLimiter
 from ddtrace.internal.remoteconfig import PayloadType
+from ddtrace.settings import _env
 from ddtrace.settings.asm import config as asm_config
 
 
@@ -72,7 +71,7 @@ def get_rules() -> str:
 
 
 def _get_rate_limiter() -> RateLimiter:
-    return RateLimiter(int(os.getenv("DD_APPSEC_TRACE_RATE_LIMIT", DEFAULT.TRACE_RATE_LIMIT)))
+    return RateLimiter(int(_env.getenv("DD_APPSEC_TRACE_RATE_LIMIT", DEFAULT.TRACE_RATE_LIMIT)))
 
 
 @dataclasses.dataclass(eq=False)

ddtrace/contrib/internal/pytest/_xdist.py

@@ -5,7 +5,6 @@
 for ITR skipping level configuration.
 """
 
-import os
 import typing as t
 
 import pytest
@@ -15,12 +14,13 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.test_visibility.api import InternalTestSession
 from ddtrace.internal.utils.formats import asbool
+from ddtrace.settings import _env
 
 
 log = get_logger(__name__)
 
 # xdist-related constants
-PYTEST_XDIST_WORKER_VALUE = os.environ.get("PYTEST_XDIST_WORKER")
+PYTEST_XDIST_WORKER_VALUE = _env.getenv("PYTEST_XDIST_WORKER")
 XDIST_UNSET = "UNSET"
 XDIST_AUTO = "auto"
 XDIST_LOGICAL = "logical"
@@ -112,7 +112,7 @@ def _skipping_level_for_xdist_parallelization_mode(
         ITR_SKIPPING_LEVEL.TEST for test-level parallelization modes (default, worksteal)
     """
     # Priority 1: Check if env var is explicitly set (not using default)
-    explicit_suite_mode = os.getenv("_DD_CIVISIBILITY_ITR_SUITE_MODE")
+    explicit_suite_mode = _env.getenv("_DD_CIVISIBILITY_ITR_SUITE_MODE")
     if explicit_suite_mode is not None:
         result = ITR_SKIPPING_LEVEL.SUITE if asbool(explicit_suite_mode) else ITR_SKIPPING_LEVEL.TEST
         log.debug(