@GitHK review: composed 'OEC:{traceback}-{timestamp}'

Author: pcrespov

packages/common-library/src/common_library/error_codes.py

@@ -10,26 +10,30 @@
 import hashlib
 import re
 import traceback
-from typing import TYPE_CHECKING, Annotated
+from datetime import datetime
+from typing import Annotated, Final, TypeAlias
 
+import arrow
 from pydantic import StringConstraints, TypeAdapter
 
-_LABEL = "OEC:{}"
-_PATTERN = r"OEC:[a-zA-Z0-9]+"
-
-if TYPE_CHECKING:
-    ErrorCodeStr = str
-else:
-    ErrorCodeStr = Annotated[
-        str, StringConstraints(strip_whitespace=True, pattern=_PATTERN)
-    ]
+_LABEL = "OEC:{fingerprint}-{timestamp}"
 
 _LEN = 12  # chars (~48 bits)
+_NAMED_PATTERN = re.compile(
+    r"OEC:(?P<fingerprint>[a-fA-F0-9]{12})-(?P<timestamp>\d{13,14})"
+    # NOTE: timestamp limits: 13 digits (from 2001), 14 digits (good for ~500+ years)
+)
+_PATTERN = re.compile(r"OEC:[a-fA-F0-9]{12}-\d{13,14}")
+
 
+ErrorCodeStr: TypeAlias = Annotated[
+    str, StringConstraints(strip_whitespace=True, pattern=_NAMED_PATTERN)
+]
 
-def _generate_error_fingerprint(exc: BaseException) -> str:
+
+def _create_fingerprint(exc: BaseException) -> str:
     """
-    Unique error fingerprint for deduplication purposes
+    Unique error fingerprint of the **traceback** for deduplication purposes
     """
     tb = traceback.extract_tb(exc.__traceback__)
     frame_sigs = [f"{frame.name}:{frame.lineno}" for frame in tb]
@@ -38,11 +42,43 @@ def _generate_error_fingerprint(exc: BaseException) -> str:
     return hashlib.sha256(fingerprint.encode()).hexdigest()[:_LEN]
 
 
+_MILISECONDS: Final[int] = 1000
+
+
+def _create_timestamp() -> int:
+    """Timestamp as milliseconds since epoch
+    NOTE: this reduces the precission to milliseconds but it is good enough for our purpose
+    """
+    ts = arrow.utcnow().float_timestamp * _MILISECONDS
+    return int(ts)
+
+
 def create_error_code(exception: BaseException) -> ErrorCodeStr:
+    """
+    Generates a unique error code for the given exception.
+
+    The error code follows the format: `OEC:{traceback}-{timestamp}`.
+    This code is intended to be shared with the front-end as a `SupportID`
+    for debugging and support purposes.
+    """
     return TypeAdapter(ErrorCodeStr).validate_python(
-        _LABEL.format(_generate_error_fingerprint(exception))
+        _LABEL.format(
+            fingerprint=_create_fingerprint(exception),
+            timestamp=_create_timestamp(),
+        )
     )
 
 
-def parse_error_code(obj) -> set[ErrorCodeStr]:
-    return set(re.findall(_PATTERN, f"{obj}"))
+def parse_error_codes(obj) -> list[ErrorCodeStr]:
+    return TypeAdapter(list[ErrorCodeStr]).validate_python(_PATTERN.findall(f"{obj}"))
+
+
+def parse_error_code_parts(oec: ErrorCodeStr) -> tuple[str, datetime]:
+    """Returns traceback-fingerprint and timestamp from `OEC:{traceback}-{timestamp}`"""
+    match = _NAMED_PATTERN.match(oec)
+    if not match:
+        msg = f"Invalid error code format: {oec}"
+        raise ValueError(msg)
+    fingerprint = match.group("fingerprint")
+    timestamp = arrow.get(int(match.group("timestamp")) / _MILISECONDS).datetime
+    return fingerprint, timestamp

packages/common-library/tests/test_error_codes.py

@@ -4,9 +4,14 @@
 # pylint: disable=unused-variable
 
 import logging
+import time
 
 import pytest
-from common_library.error_codes import create_error_code, parse_error_code
+from common_library.error_codes import (
+    create_error_code,
+    parse_error_code_parts,
+    parse_error_codes,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -31,18 +36,31 @@ def test_exception_fingerprint_consistency():
         # emulates different runs of the same function (e.g. different sessions)
         try:
             _level_one(v)  # same even if different value!
+            time.sleep(0.1)
         except Exception as err:
             error_code = create_error_code(err)
             error_codes.append(error_code)
 
-    assert error_codes == [error_codes[0]] * len(error_codes)
+    fingerprints, timestamps = list(
+        zip(
+            *[parse_error_code_parts(error_code) for error_code in error_codes],
+            strict=True,
+        )
+    )
+
+    assert fingerprints[0] == fingerprints[1]
+    assert timestamps[0] < timestamps[1]
 
+    time.sleep(0.1)
     try:
         # Same function but different location
         _level_one(0)
     except Exception as e2:
         error_code_2 = create_error_code(e2)
-        assert error_code_2 != error_code[0]
+        fingerprint_2, timestamp_2 = parse_error_code_parts(error_code_2)
+
+        assert fingerprints[0] != fingerprint_2
+        assert timestamps[1] < timestamp_2
 
 
 def test_create_log_and_parse_error_code(caplog: pytest.LogCaptureFixture):
@@ -66,11 +84,11 @@ def test_create_log_and_parse_error_code(caplog: pytest.LogCaptureFixture):
     logger.exception("Fake Unexpected error", extra={"error_code": error_code})
 
     # logs something like E.g. 2022-07-06 14:31:13,432 OEC:140350117529856 : Fake Unexpected error
-    assert parse_error_code(
+    assert parse_error_codes(
         f"2022-07-06 14:31:13,432 {error_code} : Fake Unexpected error"
-    ) == {
+    ) == [
         error_code,
-    }
+    ]
 
     assert caplog.records[0].error_code == error_code
     assert caplog.records[0]
@@ -82,6 +100,6 @@ def test_create_log_and_parse_error_code(caplog: pytest.LogCaptureFixture):
         f"This is a user-friendly message to inform about an error. [{error_code}]"
     )
 
-    assert parse_error_code(user_message) == {
+    assert parse_error_codes(user_message) == [
         error_code,
-    }
+    ]

packages/service-library/tests/aiohttp/test_rest_middlewares.py

@@ -13,7 +13,7 @@
 import pytest
 from aiohttp import web
 from aiohttp.test_utils import TestClient
-from common_library.error_codes import parse_error_code
+from common_library.error_codes import parse_error_codes
 from common_library.json_serialization import json_dumps
 from servicelib.aiohttp import status
 from servicelib.aiohttp.rest_middlewares import (
@@ -30,8 +30,7 @@ class Data:
     y: str = "foo"
 
 
-class SomeUnexpectedError(Exception):
-    ...
+class SomeUnexpectedError(Exception): ...
 
 
 class Handlers:
@@ -237,7 +236,7 @@ async def test_raised_unhandled_exception(
 
         # user friendly message with OEC reference
         assert "OEC" in error["message"]
-        parsed_oec = parse_error_code(error["message"]).pop()
+        parsed_oec = parse_error_codes(error["message"]).pop()
         assert (
             _FMSG_INTERNAL_ERROR_USER_FRIENDLY_WITH_OEC.format(error_code=parsed_oec)
             == error["message"]