Source code for multistorageclient.telemetry.attributes.base

 1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 2# SPDX-License-Identifier: Apache-2.0
 3#
 4# Licensed under the Apache License, Version 2.0 (the "License");
 5# you may not use this file except in compliance with the License.
 6# You may obtain a copy of the License at
 7#
 8# http://www.apache.org/licenses/LICENSE-2.0
 9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16from abc import ABC, abstractmethod
17from collections.abc import Sequence
18
19import opentelemetry.util.types as api_types
20
21

[docs]
22class AttributesProvider(ABC):
23    """
24    Provides :py:type:`opentelemetry.util.types.Attributes`.
25    """
26

[docs]
27    @abstractmethod
28    def attributes(self) -> api_types.Attributes:
29        """
30        Collect attributes.
31        """
32        pass


33
34
35def collect_attributes(attributes_providers: Sequence[AttributesProvider]) -> api_types.Attributes:
36    """
37    Collect and merge attributes from a sequence of attribute providers.
38
39    If multiple attributes providers return an attribute with the same key, the value from the latest attribute provider is kept.
40
41    :param attributes_providers: Attributes providers to collect attributes from.
42    :return: Merged attributes.
43    """
44    merged_attributes: api_types.Attributes = {}
45
46    for attributes in [attributes_provider.attributes() for attributes_provider in attributes_providers]:
47        if attributes is not None:
48            merged_attributes.update(attributes)
49
50    return merged_attributes