# Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 """Environment variable carriers for text map propagators. Use :class:`EnvironmentGetter` with the environment mapping to extract from, usually ``os.environ`` during application or child-process initialization. Use :class:`EnvironmentSetter` with a mutable environment copy when preparing the environment for a child process. """ import re from collections.abc import Mapping, MutableMapping from opentelemetry.propagators.textmap import Getter, Setter def _normalize_key(key: str) -> str: if not key: return "_" result = re.sub(r"[^A-Za-z0-9_]", "_", key).upper() if result and result[0].isdigit(): result = "_" + result return result def _is_normalized_key(key: str) -> bool: if not key: return False if "0" <= key[0] <= "9": return False return all( "A" <= char <= "Z" or "0" <= char <= "9" or char == "_" for char in key ) class EnvironmentGetter(Getter[Mapping[str, str]]): """Getter implementation for extracting context and baggage from environment variables. EnvironmentGetter reads from the mapping provided as the carrier, normalizes requested keys, and provides simple data access without validation. Example usage: getter = EnvironmentGetter() traceparent = getter.get(os.environ, "traceparent") """ def get(self, carrier: Mapping[str, str], key: str) -> list[str] | None: """Get a value from the environment carrier for the given key. Args: carrier: Mapping to read environment variables from key: The key to look up (will be normalized) Returns: A list with a single string value if the key exists, None otherwise. """ val = carrier.get(_normalize_key(key)) if val is None: return None return [val] def keys(self, carrier: Mapping[str, str]) -> list[str]: """Get all keys from the environment carrier. Args: carrier: Mapping to read environment variable keys from Returns: List of all already-normalized environment variable keys. """ return [key for key in carrier.keys() if _is_normalized_key(key)] class EnvironmentSetter(Setter[MutableMapping[str, str]]): """Setter implementation for building environment variable dictionaries. EnvironmentSetter builds a dictionary of environment variables that can be passed to utilities like subprocess.run() Example usage: setter = EnvironmentSetter() env_vars = {} setter.set(env_vars, "traceparent", "00-trace-id-span-id-01") subprocess.run(myCommand, env=env_vars) """ def set( self, carrier: MutableMapping[str, str], key: str, value: str ) -> None: """Set a value in the carrier dictionary for the given key. Args: carrier: Dictionary to store environment variables key: The key to set (normalized) value: The value to set """ carrier[_normalize_key(key)] = value