codekingpro/portable-devtools

1# Copyright The OpenTelemetry Authors2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15import os16import typing17from collections.abc import MutableMapping18 19from opentelemetry.propagators.textmap import Getter, Setter20 21 22class EnvironmentGetter(Getter[typing.Mapping[str, str]]):23    """Getter implementation for extracting context and baggage from environment variables.24 25    EnvironmentGetter creates a case-insensitive lookup from the current environment26    variables at initialization time and provides simple data access without validation.27 28    Per the OpenTelemetry specification, environment variables are treated as immutable29    within a process. For environments where context-carrying environment variables30    change between logical requests (e.g., AWS Lambda's _X_AMZN_TRACE_ID), create a31    new EnvironmentGetter instance at the start of each request.32 33    Example usage:34        getter = EnvironmentGetter()35        traceparent = getter.get({}, "traceparent")36    """37 38    def __init__(self):39        # Create case-insensitive lookup from current environment40        # Per spec: "creates an in-memory copy of the current environment variables"41        self.carrier: typing.Dict[str, str] = {42            k.lower(): v for k, v in os.environ.items()43        }44 45    def get(46        self, carrier: typing.Mapping[str, str], key: str47    ) -> typing.Optional[typing.List[str]]:48        """Get a value from the environment carrier for the given key.49 50        Args:51            carrier: Not used; maintained for interface compatibility with Getter[CarrierT]52            key: The key to look up (case-insensitive)53 54        Returns:55            A list with a single string value if the key exists, None otherwise.56        """57        val = self.carrier.get(key.lower())58        if val is None:59            return None60        if isinstance(val, typing.Iterable) and not isinstance(val, str):61            return list(val)62        return [val]63 64    def keys(self, carrier: typing.Mapping[str, str]) -> typing.List[str]:65        """Get all keys from the environment carrier.66 67        Args:68            carrier: Not used; maintained for interface compatibility with Getter[CarrierT]69 70        Returns:71            List of all environment variable keys (lowercase).72        """73        return list(self.carrier.keys())74 75 76class EnvironmentSetter(Setter[MutableMapping[str, str]]):77    """Setter implementation for building environment variable dictionaries.78 79    EnvironmentSetter builds a dictionary of environment variables that80    can be passed to utilities like subprocess.run()81 82    Example usage:83        setter = EnvironmentSetter()84        env_vars = {}85        setter.set(env_vars, "traceparent", "00-trace-id-span-id-01")86        subprocess.run(myCommand, env=env_vars)87    """88 89    def set(90        self, carrier: MutableMapping[str, str], key: str, value: str91    ) -> None:92        """Set a value in the carrier dictionary for the given key.93 94        Args:95            carrier: Dictionary to store environment variables96            key: The key to set (will be converted to uppercase)97            value: The value to set98        """99        carrier[key.upper()] = value100