structured json logger

cloudrunfastapi/logger.py

@@ -1,21 +1,28 @@
+import datetime
 import json
 import logging
 import os
 import socket
 from typing import Any
 
 
+class LogEncoder(json.JSONEncoder):
+    def default(self, obj: Any) -> Any:
+        if isinstance(obj, bytes):
+            return obj.decode("utf-8")
+        elif isinstance(obj, (set, frozenset)):
+            return tuple(obj)
+        elif isinstance(obj, (datetime.datetime, datetime.date, datetime.time)):
+            return obj.isoformat()
+        return super().default(obj)
+
+
 class StructuredMessage:
     def __init__(self, **kwargs: Any) -> None:
         self.kwargs = kwargs
 
     def __str__(self) -> str:
-        result = {}
-        result.update(self.kwargs)
-        return json.dumps(result)
-
-
-sm = StructuredMessage
+        return LogEncoder().encode(self.kwargs)
 
 
 class OnelineFormatter(logging.Formatter):
@@ -29,7 +36,7 @@ def format(self, record: logging.LogRecord) -> str:
             result = result.replace("\n", "")
         result_dict = record.__dict__
         result_dict["host"] = socket.gethostname()
-        return str(sm(**result_dict))
+        return str(StructuredMessage(**result_dict))
 
 
 class StructuredLogger:

cloudrunfastapi/routers/health.py

@@ -12,7 +12,8 @@
 @router.get("/healthcheck", response_model=HealthcheckResponse, tags=["health"])
 def healthcheck() -> HealthcheckResponse:
     message = "We're on the air."
-    logger.info(message)
+    time = datetime.now()
+    logger.info(msg=message, extra={"version": __version__, "time": time})
     return HealthcheckResponse(
         message=message, version=__version__, time=datetime.now()
     )

tests/test_logger.py

@@ -1,5 +1,27 @@
+from datetime import datetime
+from typing import Any
+
 from cloudrunfastapi.logger import logger
 
 
-def test_logger() -> None:
+def test_logger(capsys: Any) -> None:
     logger.exception("this is an exception")
+
+    logger.info(
+        "testing types",
+        extra={
+            "string": "string",
+            "bytes": b"test",
+            "set": {"test", "test2"},
+            "time": datetime.now(),
+        },
+    )
+
+    class MyUnsupportedType:
+        def __init__(self, data: str) -> None:
+            self.data = data
+
+    logger.info("testing types", extra={"random": MyUnsupportedType("test")})
+    out, err = capsys.readouterr()
+    assert out == ""
+    assert "TypeError: Object of type MyUnsupportedType is not JSON serializable" in err