feat(logger): type log record in LambdaPowertoolsFormatter with TypedDict

aws_lambda_powertools/logging/formatter.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import inspect
 import json
 import logging
@@ -6,10 +8,11 @@
 from abc import ABCMeta, abstractmethod
 from datetime import datetime, timezone
 from functools import partial
-from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Tuple, Union
 
 from ..shared import constants
 from ..shared.functions import powertools_dev_is_set
+from .types import PowertoolsLogRecord
 
 RESERVED_LOG_ATTRS = (
     "name",
@@ -66,7 +69,7 @@ class LambdaPowertoolsFormatter(BasePowertoolsFormatter):
 
     def __init__(
         self,
-        json_serializer: Optional[Callable[[Dict], str]] = None,
+        json_serializer: Optional[Callable[[Mapping], str]] = None,
         json_deserializer: Optional[Callable[[Union[Dict, str, bool, int, float]], str]] = None,
         json_default: Optional[Callable[[Any], Any]] = None,
         datefmt: Optional[str] = None,
@@ -144,7 +147,7 @@ def __init__(
 
         super().__init__(datefmt=self.datefmt)
 
-    def serialize(self, log: Dict) -> str:
+    def serialize(self, log: PowertoolsLogRecord | Mapping) -> str:
         """Serialize structured log dict to JSON str"""
         return self.json_serializer(log)
 

aws_lambda_powertools/logging/formatters/datadog.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
-from typing import Any, Callable
+from typing import Any, Callable, Mapping
 
 from aws_lambda_powertools.logging.formatter import LambdaPowertoolsFormatter
 
 
 class DatadogLogFormatter(LambdaPowertoolsFormatter):
     def __init__(
         self,
-        json_serializer: Callable[[dict], str] | None = None,
+        json_serializer: Callable[[Mapping], str] | None = None,
         json_deserializer: Callable[[dict | str | bool | int | float], str] | None = None,
         json_default: Callable[[Any], Any] | None = None,
         datefmt: str | None = None,

aws_lambda_powertools/logging/types.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import sys
+
+if sys.version_info < (3, 11):
+    from typing_extensions import NotRequired, TypedDict
+else:
+    from typing import NotRequired, TypedDict
+
+from typing import Any, Dict, List
+
+
+class PowertoolsLogRecord(TypedDict):
+    # Base fields (required)
+    level: str
+    location: str
+    message: Dict[str, Any] | str | bool | List[Any]
+    timestamp: str | int
+    service: str
+
+    # Fields from logger.inject_lambda_context
+    cold_start: NotRequired[bool]
+    function_name: NotRequired[str]
+    function_memory_size: NotRequired[int]
+    function_arn: NotRequired[str]
+    function_request_id: NotRequired[str]
+    # From logger.inject_lambda_context if AWS X-Ray is enabled
+    xray_trace_id: NotRequired[str]
+
+    # If sample_rate is defined
+    sampling_rate: NotRequired[float]
+
+    # From logger.set_correlation_id
+    correlation_id: NotRequired[str]
+
+    # Fields from logger.exception
+    exception_name: NotRequired[str]
+    exception: NotRequired[str]

examples/logger/src/bring_your_own_formatter.py

@@ -1,12 +1,17 @@
+from __future__ import annotations
+
+from typing import Mapping
+
 from aws_lambda_powertools import Logger
 from aws_lambda_powertools.logging.formatter import LambdaPowertoolsFormatter
+from aws_lambda_powertools.logging.types import PowertoolsLogRecord
 
 
 class CustomFormatter(LambdaPowertoolsFormatter):
-    def serialize(self, log: dict) -> str:
+    def serialize(self, log: PowertoolsLogRecord | Mapping) -> str:  
         """Serialize final structured log dict to JSON str"""
-        log["event"] = log.pop("message")  # rename message key to event
-        return self.json_serializer(log)  # use configured json serializer
+        log["event"] = log.pop("message")  # type: ignore
+        return self.json_serializer(log)  
 
 
 logger = Logger(service="payment", logger_formatter=CustomFormatter())