logs: introduce LogAttributes type

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add proper type for logs signal attributes
+  ([#4342](https://github.com/open-telemetry/opentelemetry-python/pull/4342))
+
 ## Version 1.29.0/0.50b0 (2024-12-11)
 
 - Fix crash exporting a log record with None body

opentelemetry-api/src/opentelemetry/_logs/_internal/__init__.py

@@ -37,14 +37,14 @@
 from logging import getLogger
 from os import environ
 from time import time_ns
-from typing import Any, Optional, cast
+from typing import Optional, cast
 
 from opentelemetry._logs.severity import SeverityNumber
 from opentelemetry.environment_variables import _OTEL_PYTHON_LOGGER_PROVIDER
 from opentelemetry.trace.span import TraceFlags
 from opentelemetry.util._once import Once
 from opentelemetry.util._providers import _load_provider
-from opentelemetry.util.types import Attributes
+from opentelemetry.util.types import AnyValue, LogAttributes
 
 _logger = getLogger(__name__)
 
@@ -66,8 +66,8 @@ def __init__(
         trace_flags: Optional["TraceFlags"] = None,
         severity_text: Optional[str] = None,
         severity_number: Optional[SeverityNumber] = None,
-        body: Optional[Any] = None,
-        attributes: Optional["Attributes"] = None,
+        body: AnyValue = None,
+        attributes: LogAttributes = None,
     ):
         self.timestamp = timestamp
         if observed_timestamp is None:
@@ -78,7 +78,7 @@ def __init__(
         self.trace_flags = trace_flags
         self.severity_text = severity_text
         self.severity_number = severity_number
-        self.body = body  # type: ignore
+        self.body = body
         self.attributes = attributes
 
 
@@ -90,7 +90,7 @@ def __init__(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> None:
         super().__init__()
         self._name = name
@@ -119,7 +119,7 @@ def __init__(  # pylint: disable=super-init-not-called
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ):
         self._name = name
         self._version = version
@@ -158,7 +158,7 @@ def get_logger(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> Logger:
         """Returns a `Logger` for use by the given instrumentation library.
 
@@ -196,7 +196,7 @@ def get_logger(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> Logger:
         """Returns a NoOpLogger."""
         return NoOpLogger(
@@ -210,7 +210,7 @@ def get_logger(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> Logger:
         if _LOGGER_PROVIDER:
             return _LOGGER_PROVIDER.get_logger(
@@ -273,7 +273,7 @@ def get_logger(
     instrumenting_library_version: str = "",
     logger_provider: Optional[LoggerProvider] = None,
     schema_url: Optional[str] = None,
-    attributes: Optional[Attributes] = None,
+    attributes: LogAttributes = None,
 ) -> "Logger":
     """Returns a `Logger` for use within a python process.
 

opentelemetry-api/src/opentelemetry/util/types.py

@@ -56,3 +56,5 @@
     ],
     ...,
 ]
+
+LogAttributes = Optional[Mapping[str, "AnyValue"]]

opentelemetry-api/tests/logs/test_proxy.py

@@ -19,7 +19,7 @@
 import opentelemetry._logs._internal as _logs_internal
 from opentelemetry import _logs
 from opentelemetry.test.globals_test import LoggingGlobalsTest
-from opentelemetry.util.types import Attributes
+from opentelemetry.util.types import LogAttributes
 
 
 class TestProvider(_logs.NoOpLoggerProvider):
@@ -28,7 +28,7 @@ def get_logger(
         name: str,
         version: typing.Optional[str] = None,
         schema_url: typing.Optional[str] = None,
-        attributes: typing.Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> _logs.Logger:
         return LoggerTest(name)
 

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py

@@ -51,7 +51,7 @@
     get_current_span,
 )
 from opentelemetry.trace.span import TraceFlags
-from opentelemetry.util.types import AnyValue, Attributes
+from opentelemetry.util.types import AnyValue, LogAttributes
 
 _logger = logging.getLogger(__name__)
 
@@ -179,9 +179,9 @@ def __init__(
         trace_flags: Optional[TraceFlags] = None,
         severity_text: Optional[str] = None,
         severity_number: Optional[SeverityNumber] = None,
-        body: Optional[AnyValue] = None,
+        body: AnyValue = None,
         resource: Optional[Resource] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
         limits: Optional[LogLimits] = _UnsetLogLimits,
     ):
         super().__init__(
@@ -476,7 +476,7 @@ def __init__(
         self._logger_provider = logger_provider or get_logger_provider()
 
     @staticmethod
-    def _get_attributes(record: logging.LogRecord) -> Attributes:
+    def _get_attributes(record: logging.LogRecord) -> LogAttributes:
         attributes = {
             k: v for k, v in vars(record).items() if k not in _RESERVED_ATTRS
         }
@@ -633,7 +633,7 @@ def _get_logger_no_cache(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> Logger:
         return Logger(
             self._resource,
@@ -667,7 +667,7 @@ def get_logger(
         name: str,
         version: Optional[str] = None,
         schema_url: Optional[str] = None,
-        attributes: Optional[Attributes] = None,
+        attributes: LogAttributes = None,
     ) -> Logger:
         if self._disabled:
             warnings.warn("SDK is disabled.")