Small changes

Author: leandrodamascena

aws_lambda_powertools/logging/formatter.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import inspect
-import sys
 import json
 import logging
 import os
@@ -11,11 +10,8 @@
 from datetime import datetime, timezone
 from functools import partial
 from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
-from ..shared.functions import (
-    resolve_truthy_env_var_choice
-)
 
-from aws_lambda_powertools.logging.types import LogRecord
+from aws_lambda_powertools.logging.types import LogRecord, LogStackTrace
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import powertools_dev_is_set
 
@@ -161,39 +157,21 @@ def __init__(
     def serialize(self, log: LogRecord) -> str:
         """Serialize structured log dict to JSON str"""
         return self.json_serializer(log)
-    
-    def serialize_traceback(self, log_record: logging.LogRecord) -> list:
-        exception_info = {
-            "type": log_record.exc_info[0].__name__,
-            "value": log_record.exc_info[1],
-            "module": log_record.exc_info[1].__class__.__module__,
-            "frames": []
-        }
-
-        exception_info["frames"] = [{
-                "file": fs.filename,
-                "line": fs.lineno,
-                "function": fs.name,
-                "statement": fs.line
-            } for fs in traceback.extract_tb(log_record.exc_info[2])]
-
-        return exception_info
 
     def format(self, record: logging.LogRecord) -> str:  # noqa: A003
         """Format logging record as structured JSON str"""
         formatted_log = self._extract_log_keys(log_record=record)
         formatted_log["message"] = self._extract_log_message(log_record=record)
 
-        if self.serialize_stacktrace:
-            # Generate the traceback from the traceback library
-            formatted_log["stack_trace"] = self.serialize_traceback(log_record=record)
-
         # exception and exception_name fields can be added as extra key
         # in any log level, we try to extract and use them first
         extracted_exception, extracted_exception_name = self._extract_log_exception(log_record=record)
         formatted_log["record"] = record
         formatted_log["exception"] = formatted_log.get("exception", extracted_exception)
         formatted_log["exception_name"] = formatted_log.get("exception_name", extracted_exception_name)
+        if self.serialize_stacktrace:
+            # Generate the traceback from the traceback library
+            formatted_log["stack_trace"] = self._serialize_stacktrace(log_record=record)
         formatted_log["xray_trace_id"] = self._get_latest_trace_id()
         formatted_log = self._strip_none_records(records=formatted_log)
 
@@ -304,6 +282,24 @@ def _extract_log_message(self, log_record: logging.LogRecord) -> Union[Dict[str,
 
         return message
 
+    def _serialize_stacktrace(self, log_record: logging.LogRecord) -> LogStackTrace | None:
+        if log_record.exc_info:
+            exception_info: LogStackTrace = {
+                "type": log_record.exc_info[0].__name__,  # type: ignore
+                "value": log_record.exc_info[1],  # type: ignore
+                "module": log_record.exc_info[1].__class__.__module__,
+                "frames": [],
+            }
+
+            exception_info["frames"] = [
+                {"file": fs.filename, "line": fs.lineno, "function": fs.name, "statement": fs.line}
+                for fs in traceback.extract_tb(log_record.exc_info[2])
+            ]
+
+            return exception_info
+
+        return None
+
     def _extract_log_exception(self, log_record: logging.LogRecord) -> Union[Tuple[str, str], Tuple[None, None]]:
         """Format traceback information, if available
 

aws_lambda_powertools/logging/logger.py

@@ -254,7 +254,7 @@ def __init__(
             "log_record_order": log_record_order,
             "utc": utc,
             "use_rfc3339": use_rfc3339,
-            "serialize_stacktrace": serialize_stacktrace
+            "serialize_stacktrace": serialize_stacktrace,
         }
 
         self._init_logger(formatter_options=formatter_options, log_level=level, **kwargs)

aws_lambda_powertools/logging/types.py

@@ -5,6 +5,7 @@
 from aws_lambda_powertools.shared.types import NotRequired, TypeAlias, TypedDict
 
 LogRecord: TypeAlias = Union[Dict[str, Any], "PowertoolsLogRecord"]
+LogStackTrace: TypeAlias = Union[Dict[str, Any], "PowertoolsStackTrace"]
 
 
 class PowertoolsLogRecord(TypedDict):
@@ -33,3 +34,11 @@ class PowertoolsLogRecord(TypedDict):
     # Fields from logger.exception
     exception_name: NotRequired[str]
     exception: NotRequired[str]
+    stack_trace: NotRequired[Dict[str, Any]]
+
+
+class PowertoolsStackTrace(TypedDict):
+    type: str
+    value: str
+    module: str
+    frames: List[Dict[str, Any]]

docs/contributing/documentation/rfcs.md

@@ -3,6 +3,8 @@ title: Writing Request For Comment (RFC)
 description: Contributing RFCs to Powertools for AWS Lambda (Python)
 ---
 
+<!-- markdownlint-disable MD043 -->
+
 ## TBW
 
 Something great will come.

examples/logger/src/logging_stacktrace.py

@@ -1,8 +1,8 @@
 import requests
 
 from aws_lambda_powertools import Logger
-from aws_lambda_powertools.utilities.typing import LambdaContext
 from aws_lambda_powertools.logging.formatter import LambdaPowertoolsFormatter
+from aws_lambda_powertools.utilities.typing import LambdaContext
 
 ENDPOINT = "http://httpbin.org/status/500"
 logger = Logger(logger_formatter=LambdaPowertoolsFormatter(serialize_stacktrace=True))

tests/functional/test_logger.py

@@ -960,6 +960,7 @@ def test_stream_defaults_to_stdout(service_name, capsys):
     log = json.loads(capsys.readouterr().out.strip())
     assert log["message"] == msg
 
+
 def test_logger_logs_stack_trace_without_parameter(stdout):
     logger = Logger(stream=stdout)
 
@@ -971,6 +972,7 @@ def test_logger_logs_stack_trace_without_parameter(stdout):
     log = capture_logging_output(stdout)
     assert "stack_trace" in log
 
+
 def test_logger_logs_stack_trace_with_parameter(stdout):
     logger = Logger(stream=stdout, logger_formatter=LambdaPowertoolsFormatter(serialize_stacktrace=True))
 
@@ -982,6 +984,7 @@ def test_logger_logs_stack_trace_with_parameter(stdout):
     log = capture_logging_output(stdout)
     assert "stack_trace" in log
 
+
 def test_logger_logs_no_stack_trace_with_parameter_false(stdout):
     logger = Logger(stream=stdout, logger_formatter=LambdaPowertoolsFormatter(serialize_stacktrace=False))
 
@@ -993,6 +996,7 @@ def test_logger_logs_no_stack_trace_with_parameter_false(stdout):
     log = capture_logging_output(stdout)
     assert "stack_trace" not in log
 
+
 def test_logger_logs_stack_trace_with_logger_parameter(stdout):
     logger = Logger(stream=stdout, serialize_stacktrace=True)
 
@@ -1004,6 +1008,7 @@ def test_logger_logs_stack_trace_with_logger_parameter(stdout):
     log = capture_logging_output(stdout)
     assert "stack_trace" in log.keys()
 
+
 def test_logger_logs_no_stack_trace_with_logger_parameter_false(stdout):
     logger = Logger(stream=stdout, serialize_stacktrace=False)