Added structured logging using Python logging to mimic GCP logs

changelog_entry.yaml

@@ -0,0 +1,4 @@
+- bump: patch
+  changes:
+    added:
+    - Implemented structured logging in policyengine_api/utils/logger.py to produce GCP-compatible JSON logs.

policyengine_api/libs/simulation_api.py

@@ -5,7 +5,7 @@
 import time
 from typing import Any, Literal, Annotated
 from dotenv import load_dotenv
-from policyengine_api.gcp_logging import logger
+from policyengine_api.utils.python_logging import logger
 from google.cloud.workflows import executions_v1
 
 
@@ -21,7 +21,7 @@ class SimulationAPI:
 
     def __init__(self):
         if os.environ.get("GOOGLE_APPLICATION_CREDENTIALS") is None:
-            logger.log_text(
+            logger.log_struct(
                 "GOOGLE_APPLICATION_CREDENTIALS not set; unable to run simulation API.",
                 severity="ERROR",
             )

policyengine_api/services/economy_service.py

@@ -3,7 +3,7 @@
     ReformImpactsService,
 )
 from policyengine_api.constants import COUNTRY_PACKAGE_VERSIONS
-from policyengine_api.gcp_logging import logger
+from policyengine_api.utils.python_logging import logger
 from policyengine_api.libs.simulation_api import SimulationAPI
 from policyengine_api.data.model_setup import get_dataset_version
 from policyengine.simulation import SimulationOptions

policyengine_api/utils/logger.py

@@ -0,0 +1,49 @@
+import logging
+import json
+import sys
+from datetime import datetime, timezone
+
+
+class Logger:
+    def __init__(self, name="policyengine-api", level=logging.INFO):
+        self.logger = logging.getLogger(name)
+        self.logger.setLevel(level)
+
+        if not self.logger.handlers:
+            handler = logging.StreamHandler(sys.stdout)
+            handler.setFormatter(logging.Formatter("%(message)s"))
+            self.logger.addHandler(handler)
+
+    def log_struct(
+        self, payload: dict, severity: str = "INFO", message: str = None
+    ):
+        log_entry = {
+            "severity": severity.upper(),
+            "timestamp": datetime.now(timezone.utc)
+            .isoformat(timespec="milliseconds")
+            .replace("+00:00", "Z"),
+        }
+
+        if message:
+            log_entry["message"] = message
+
+        if isinstance(payload, dict):
+            log_entry.update(payload)
+
+        json_log = json.dumps(log_entry, ensure_ascii=False)
+        self._emit(severity, json_log)
+
+    def _emit(self, severity: str, msg: str):
+        level = severity.upper()
+        if level == "DEBUG":
+            self.logger.debug(msg)
+        elif level == "INFO":
+            self.logger.info(msg)
+        elif level == "WARNING":
+            self.logger.warning(msg)
+        elif level == "ERROR":
+            self.logger.error(msg)
+        elif level == "CRITICAL":
+            self.logger.critical(msg)
+        else:
+            self.logger.info(msg)

policyengine_api/utils/python_logging.py

@@ -0,0 +1,3 @@
+from policyengine_api.utils.logger import Logger
+
+logger = Logger("policyengine-api")

tests/unit/test_logger.py

@@ -0,0 +1,14 @@
+from policyengine_api.utils.logger import Logger
+
+
+def test_log_struct(capsys):
+    logger = Logger("test-logger")
+    logger.log_struct(
+        {"event": "test_event", "user_id": "abc123"},
+        severity="INFO",
+        message="Logging from test",
+    )
+    captured = capsys.readouterr()
+    assert '"severity": "INFO"' in captured.out
+    assert '"event": "test_event"' in captured.out
+    assert '"message": "Logging from test"' in captured.out