Chore: Updated references using pyupgrade

Author: jomcgi

docs/examples/metrics/instruments/example.py

@@ -1,4 +1,4 @@
-from typing import Iterable
+from collections.abc import Iterable
 
 from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
     OTLPMetricExporter,

docs/examples/metrics/views/change_reservoir_factory.py

@@ -14,7 +14,6 @@
 
 import random
 import time
-from typing import Type
 
 from opentelemetry import trace
 from opentelemetry.metrics import get_meter_provider, set_meter_provider
@@ -39,7 +38,7 @@
 
 # Create a custom reservoir factory with specified parameters
 def custom_reservoir_factory(
-    aggregationType: Type[_Aggregation],
+    aggregationType: type[_Aggregation],
 ) -> ExemplarReservoirBuilder:
     if issubclass(aggregationType, _ExplicitBucketHistogramAggregation):
         return AlignedHistogramBucketExemplarReservoir

docs/examples/metrics/views/limit_num_of_attrs.py

@@ -14,7 +14,7 @@
 
 import random
 import time
-from typing import Iterable
+from collections.abc import Iterable
 
 from opentelemetry.metrics import (
     CallbackOptions,

docs/getting_started/metrics_example.py

@@ -15,7 +15,7 @@
 # metrics.py
 # This is still work in progress as the metrics SDK is being implemented
 
-from typing import Iterable
+from collections.abc import Iterable
 
 from opentelemetry.metrics import (
     CallbackOptions,

exporter/opentelemetry-exporter-opencensus/src/opentelemetry/exporter/opencensus/trace_exporter/__init__.py

@@ -15,7 +15,7 @@
 """OpenCensus Span Exporter."""
 
 import logging
-from typing import Sequence
+from collections.abc import Sequence
 
 import grpc
 from opencensus.proto.agent.trace.v1 import (

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/__init__.py

@@ -16,16 +16,11 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from itertools import count
 from typing import (
     Any,
     Callable,
-    Dict,
-    Iterator,
-    List,
-    Mapping,
-    Optional,
     TypeVar,
 )
 
@@ -69,9 +64,7 @@ def _encode_resource(resource: Resource) -> PB2Resource:
     return PB2Resource(attributes=_encode_attributes(resource.attributes))
 
 
-def _encode_value(
-    value: Any, allow_null: bool = False
-) -> Optional[PB2AnyValue]:
+def _encode_value(value: Any, allow_null: bool = False) -> PB2AnyValue | None:
     if allow_null is True and value is None:
         return None
     if isinstance(value, bool):
@@ -137,7 +130,7 @@ def _encode_trace_id(trace_id: int) -> bytes:
 
 def _encode_attributes(
     attributes: Attributes,
-) -> Optional[List[PB2KeyValue]]:
+) -> list[PB2KeyValue] | None:
     if attributes:
         pb2_attributes = []
         for key, value in attributes.items():
@@ -152,10 +145,10 @@ def _encode_attributes(
 
 
 def _get_resource_data(
-    sdk_resource_scope_data: Dict[Resource, _ResourceDataT],
+    sdk_resource_scope_data: dict[Resource, _ResourceDataT],
     resource_class: Callable[..., _TypingResourceT],
     name: str,
-) -> List[_TypingResourceT]:
+) -> list[_TypingResourceT]:
     resource_data = []
 
     for (
@@ -169,7 +162,7 @@ def _get_resource_data(
             resource_class(
                 **{
                     "resource": collector_resource,
-                    "scope_{}".format(name): scope_data.values(),
+                    f"scope_{name}": scope_data.values(),
                 }
             )
         )

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/_log_encoder/__init__.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from collections import defaultdict
-from typing import List, Sequence
+from collections.abc import Sequence
 
 from opentelemetry.exporter.otlp.proto.common._internal import (
     _encode_attributes,
@@ -63,7 +63,7 @@ def _encode_log(log_data: LogData) -> PB2LogRecord:
     )
 
 
-def _encode_resource_logs(batch: Sequence[LogData]) -> List[ResourceLogs]:
+def _encode_resource_logs(batch: Sequence[LogData]) -> list[ResourceLogs]:
     sdk_resource_logs = defaultdict(lambda: defaultdict(list))
 
     for sdk_log in batch:

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/metrics_encoder/__init__.py

@@ -15,7 +15,6 @@
 
 import logging
 from os import environ
-from typing import Dict, List
 
 from opentelemetry.exporter.otlp.proto.common._internal import (
     _encode_attributes,
@@ -79,8 +78,8 @@ def _common_configuration(
         )
 
     def _get_temporality(
-        self, preferred_temporality: Dict[type, AggregationTemporality]
-    ) -> Dict[type, AggregationTemporality]:
+        self, preferred_temporality: dict[type, AggregationTemporality]
+    ) -> dict[type, AggregationTemporality]:
         otel_exporter_otlp_metrics_temporality_preference = (
             environ.get(
                 OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
@@ -135,8 +134,8 @@ def _get_temporality(
 
     def _get_aggregation(
         self,
-        preferred_aggregation: Dict[type, Aggregation],
-    ) -> Dict[type, Aggregation]:
+        preferred_aggregation: dict[type, Aggregation],
+    ) -> dict[type, Aggregation]:
         otel_exporter_otlp_metrics_default_histogram_aggregation = environ.get(
             OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION,
             "explicit_bucket_histogram",
@@ -343,7 +342,7 @@ def _encode_metric(metric, pb2_metric):
         )
 
 
-def _encode_exemplars(sdk_exemplars: List[Exemplar]) -> List[pb2.Exemplar]:
+def _encode_exemplars(sdk_exemplars: list[Exemplar]) -> list[pb2.Exemplar]:
     """
     Converts a list of SDK Exemplars into a list of protobuf Exemplars.
 

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/trace_encoder/__init__.py

@@ -14,7 +14,8 @@
 
 import logging
 from collections import defaultdict
-from typing import List, Optional, Sequence
+from collections.abc import Sequence
+from typing import Optional
 
 from opentelemetry.exporter.otlp.proto.common._internal import (
     _encode_attributes,
@@ -59,7 +60,7 @@ def encode_spans(
 
 def _encode_resource_spans(
     sdk_spans: Sequence[ReadableSpan],
-) -> List[PB2ResourceSpans]:
+) -> list[PB2ResourceSpans]:
     # We need to inspect the spans and group + structure them as:
     #
     #   Resource
@@ -136,7 +137,7 @@ def _encode_span(sdk_span: ReadableSpan) -> PB2SPan:
 
 def _encode_events(
     events: Sequence[Event],
-) -> Optional[List[PB2SPan.Event]]:
+) -> Optional[list[PB2SPan.Event]]:
     pb2_events = None
     if events:
         pb2_events = []

exporter/opentelemetry-exporter-otlp-proto-common/tests/test_log_encoder.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import unittest
-from typing import List, Tuple
 
 from opentelemetry._logs import SeverityNumber
 from opentelemetry.exporter.otlp.proto.common._internal import (
@@ -83,7 +82,7 @@ def test_dropped_attributes_count(self):
         )
 
     @staticmethod
-    def _get_sdk_log_data() -> List[LogData]:
+    def _get_sdk_log_data() -> list[LogData]:
         log1 = LogData(
             log_record=SDKLogRecord(
                 timestamp=1644650195189786880,
@@ -229,7 +228,7 @@ def _get_sdk_log_data() -> List[LogData]:
 
     def get_test_logs(
         self,
-    ) -> Tuple[List[SDKLogRecord], ExportLogsServiceRequest]:
+    ) -> tuple[list[SDKLogRecord], ExportLogsServiceRequest]:
         sdk_logs = self._get_sdk_log_data()
 
         pb2_service_request = ExportLogsServiceRequest(
@@ -479,7 +478,7 @@ def get_test_logs(
         return sdk_logs, pb2_service_request
 
     @staticmethod
-    def _get_test_logs_dropped_attributes() -> List[LogData]:
+    def _get_test_logs_dropped_attributes() -> list[LogData]:
         log1 = LogData(
             log_record=SDKLogRecord(
                 timestamp=1644650195189786880,

exporter/opentelemetry-exporter-otlp-proto-common/tests/test_trace_encoder.py

@@ -15,7 +15,6 @@
 # pylint: disable=protected-access
 
 import unittest
-from typing import List, Tuple
 
 from opentelemetry.exporter.otlp.proto.common._internal import (
     _encode_span_id,
@@ -63,7 +62,7 @@ def test_encode_spans(self):
         self.assertEqual(encode_spans(otel_spans), expected_encoding)
 
     @staticmethod
-    def get_exhaustive_otel_span_list() -> List[SDKSpan]:
+    def get_exhaustive_otel_span_list() -> list[SDKSpan]:
         trace_id = 0x3E0C63257DE34C926F9EFCD03927272E
 
         base_time = 683647322 * 10**9  # in ns
@@ -194,7 +193,7 @@ def get_exhaustive_otel_span_list() -> List[SDKSpan]:
 
     def get_exhaustive_test_spans(
         self,
-    ) -> Tuple[List[SDKSpan], PB2ExportTraceServiceRequest]:
+    ) -> tuple[list[SDKSpan], PB2ExportTraceServiceRequest]:
         otel_spans = self.get_exhaustive_otel_span_list()
         trace_id = _encode_trace_id(otel_spans[0].context.trace_id)
         span_kind = _SPAN_KIND_MAP[SDKSpanKind.INTERNAL]

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/_log_exporter/__init__.py

@@ -11,9 +11,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from collections.abc import Sequence
+from collections.abc import Sequence as TypingSequence
 from os import environ
-from typing import Dict, Optional, Sequence, Tuple, Union
-from typing import Sequence as TypingSequence
+from typing import Optional, Union
 
 from grpc import ChannelCredentials, Compression
 from opentelemetry.exporter.otlp.proto.common._log_encoder import encode_logs
@@ -56,7 +57,7 @@ def __init__(
         insecure: Optional[bool] = None,
         credentials: Optional[ChannelCredentials] = None,
         headers: Optional[
-            Union[TypingSequence[Tuple[str, str]], Dict[str, str], str]
+            Union[TypingSequence[tuple[str, str]], dict[str, str], str]
         ] = None,
         timeout: Optional[int] = None,
         compression: Optional[Compression] = None,

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py

@@ -17,21 +17,18 @@
 import threading
 from abc import ABC, abstractmethod
 from collections.abc import Sequence  # noqa: F401
+from collections.abc import Sequence as TypingSequence
 from logging import getLogger
 from os import environ
 from time import sleep
 from typing import (  # noqa: F401
     Any,
     Callable,
-    Dict,
     Generic,
-    List,
     Optional,
-    Tuple,
     TypeVar,
     Union,
 )
-from typing import Sequence as TypingSequence
 from urllib.parse import urlparse
 
 from deprecated import deprecated
@@ -112,10 +109,10 @@ def environ_to_compression(environ_key: str) -> Optional[Compression]:
     reason="Use one of the encoders from opentelemetry-exporter-otlp-proto-common instead",
 )
 def get_resource_data(
-    sdk_resource_scope_data: Dict[SDKResource, ResourceDataT],
+    sdk_resource_scope_data: dict[SDKResource, ResourceDataT],
     resource_class: Callable[..., TypingResourceT],
     name: str,
-) -> List[TypingResourceT]:
+) -> list[TypingResourceT]:
     return _get_resource_data(sdk_resource_scope_data, resource_class, name)
 
 
@@ -192,7 +189,7 @@ def __init__(
         insecure: Optional[bool] = None,
         credentials: Optional[ChannelCredentials] = None,
         headers: Optional[
-            Union[TypingSequence[Tuple[str, str]], Dict[str, str], str]
+            Union[TypingSequence[tuple[str, str]], dict[str, str], str]
         ] = None,
         timeout: Optional[int] = None,
         compression: Optional[Compression] = None,

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/metric_exporter/__init__.py

@@ -13,11 +13,11 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable
+from collections.abc import Sequence as TypingSequence
 from dataclasses import replace
 from logging import getLogger
 from os import environ
-from typing import Iterable, List, Tuple, Union
-from typing import Sequence as TypingSequence
 
 from grpc import ChannelCredentials, Compression
 from opentelemetry.exporter.otlp.proto.common._internal.metrics_encoder import (
@@ -97,7 +97,9 @@ def __init__(
         endpoint: str | None = None,
         insecure: bool | None = None,
         credentials: ChannelCredentials | None = None,
-        headers: Union[TypingSequence[Tuple[str, str]], dict[str, str], str]
+        headers: TypingSequence[tuple[str, str]]
+        | dict[str, str]
+        | str
         | None = None,
         timeout: int | None = None,
         compression: Compression | None = None,
@@ -180,26 +182,26 @@ def _split_metrics_data(
         metrics_data: MetricsData,
     ) -> Iterable[MetricsData]:
         batch_size: int = 0
-        split_resource_metrics: List[ResourceMetrics] = []
+        split_resource_metrics: list[ResourceMetrics] = []
 
         for resource_metrics in metrics_data.resource_metrics:
-            split_scope_metrics: List[ScopeMetrics] = []
+            split_scope_metrics: list[ScopeMetrics] = []
             split_resource_metrics.append(
                 replace(
                     resource_metrics,
                     scope_metrics=split_scope_metrics,
                 )
             )
             for scope_metrics in resource_metrics.scope_metrics:
-                split_metrics: List[Metric] = []
+                split_metrics: list[Metric] = []
                 split_scope_metrics.append(
                     replace(
                         scope_metrics,
                         metrics=split_metrics,
                     )
                 )
                 for metric in scope_metrics.metrics:
-                    split_data_points: List[DataPointT] = []
+                    split_data_points: list[DataPointT] = []
                     split_metrics.append(
                         replace(
                             metric,