Add 'datetime_fmt' for log message timestamps using datetime

Author: huonw

aws_lambda_powertools/logging/formatter.py

@@ -3,6 +3,7 @@
 import os
 import time
 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
 
@@ -61,9 +62,10 @@ def __init__(
         json_deserializer: Optional[Callable[[Union[Dict, str, bool, int, float]], str]] = None,
         json_default: Optional[Callable[[Any], Any]] = None,
         datefmt: Optional[str] = None,
+        datetime_fmt: Optional[str] = None,
         log_record_order: Optional[List[str]] = None,
         utc: bool = False,
-        **kwargs
+        **kwargs,
     ):
         """Return a LambdaPowertoolsFormatter instance.
 
@@ -86,20 +88,33 @@ def __init__(
             Only used when no custom JSON encoder is set
 
         datefmt : str, optional
-            String directives (strftime) to format log timestamp
+            String directives (strftime) to format log timestamp using `time`. Only one of `datefmt`
+            and `datetime_fmt` should be specified.
 
             See https://docs.python.org/3/library/time.html#time.strftime
+        datetime_fmt : str, optional
+            String directives (strftime) to format log timestamp using `datetime`. Only one of
+            `datefmt` and `datetime_fmt` should be specified.
+
+            See https://docs.python.org/3/library/datetime.html#strftime-strptime-behavior . This
+            also supports a custom %F directive for milliseconds.
         utc : bool, optional
             set logging timestamp to UTC, by default False to continue to use local time as per stdlib
         log_record_order : list, optional
             set order of log keys when logging, by default ["level", "location", "message", "timestamp"]
         kwargs
             Key-value to be included in log messages
+
         """
         self.json_deserializer = json_deserializer or json.loads
         self.json_default = json_default or str
         self.json_serializer = json_serializer or partial(json.dumps, default=self.json_default, separators=(",", ":"))
+
+        if datefmt and datetime_fmt:
+            raise ValueError(f"at most one of datefmt {datefmt!r} and datetime_fmt {datetime_fmt!r} can be specified")
         self.datefmt = datefmt
+        self.datetime_fmt = datetime_fmt
+
         self.utc = utc
         self.log_record_order = log_record_order or ["level", "location", "message", "timestamp"]
         self.log_format = dict.fromkeys(self.log_record_order)  # Set the insertion order for the log messages
@@ -129,13 +144,26 @@ def format(self, record: logging.LogRecord) -> str:  # noqa: A003
 
     def formatTime(self, record: logging.LogRecord, datefmt: Optional[str] = None) -> str:
         record_ts = self.converter(record.created)  # type: ignore
-        if datefmt:
-            return time.strftime(datefmt, record_ts)
 
         # NOTE: Python `time.strftime` doesn't provide msec directives
         # so we create a custom one (%F) and replace logging record ts
         # Reason 2 is that std logging doesn't support msec after TZ
         msecs = "%03d" % record.msecs
+
+        if datefmt or self.datefmt:
+            return time.strftime(datefmt or self.datefmt, record_ts)
+
+        elif self.datetime_fmt:
+            timestamp = record.created + record.msecs / 1000
+
+            dt = datetime.fromtimestamp(timestamp, tz=timezone.utc)
+            if not self.utc:
+                # convert back to local time
+                dt = dt.astimezone()
+
+            custom_fmt = self.datetime_fmt.replace(self.custom_ms_time_directive, msecs)
+            return dt.strftime(custom_fmt)
+
         custom_fmt = self.default_time_format.replace(self.custom_ms_time_directive, msecs)
         return time.strftime(custom_fmt, record_ts)
 
@@ -219,7 +247,7 @@ def _extract_log_keys(self, log_record: logging.LogRecord) -> Dict[str, Any]:
             Structured log as dictionary
         """
         record_dict = log_record.__dict__.copy()
-        record_dict["asctime"] = self.formatTime(record=log_record, datefmt=self.datefmt)
+        record_dict["asctime"] = self.formatTime(record=log_record)
         extras = {k: v for k, v in record_dict.items() if k not in RESERVED_LOG_ATTRS}
 
         formatted_log = {}

aws_lambda_powertools/logging/logger.py

@@ -80,7 +80,11 @@ class Logger(logging.Logger):  # lgtm [py/missing-call-to-init]
     Parameters propagated to LambdaPowertoolsFormatter
     ---------------------------------------------
     datefmt: str, optional
-        String directives (strftime) to format log timestamp, by default it uses RFC 3339.
+        String directives (strftime) to format log timestamp using `time`, by default it uses RFC
+        3339. Only one of `datefmt` and `datetime_fmt` should be specified.
+    datetime_fmt : str, optional
+        String directives (strftime) to format log timestamp using `datetime`. Only one of `datefmt`
+        and `datetime_fmt` should be specified.
     json_serializer : Callable, optional
         function to serialize `obj` to a JSON formatted `str`, by default json.dumps
     json_deserializer : Callable, optional

tests/functional/test_logger.py

@@ -3,8 +3,10 @@
 import json
 import logging
 import random
+import re
 import string
 from collections import namedtuple
+from datetime import datetime, timezone
 from typing import Iterable
 
 import pytest
@@ -610,3 +612,25 @@ def handler(event, context, my_custom_option=None):
 
     # THEN
     handler({}, lambda_context, my_custom_option="blah")
+
+
+@pytest.mark.parametrize("utc", [False, True])
+def test_datetime_fmt(stdout, service_name, utc):
+    # GIVEN
+    logger = Logger(
+        service=service_name,
+        stream=stdout,
+        datetime_fmt="custom timestamp: milliseconds=%F microseconds=%f timezone=%z",
+        utc=utc,
+    )
+
+    # WHEN a log statement happens
+    logger.info({})
+
+    # THEN the timestamp has the appropriate formatting
+    log = capture_logging_output(stdout)
+
+    expected_tz = datetime.now().astimezone(timezone.utc if utc else None).strftime("%z")
+    assert re.fullmatch(
+        f"custom timestamp: milliseconds=[0-9]+ microseconds=[0-9]+ timezone={re.escape(expected_tz)}", log["timestamp"]
+    )