change: Define workflow data as a dict and use dict_to_logfmt

Author: duhow

src/app.py

@@ -7,7 +7,7 @@
 
 
 from const import GithubHeaders, LOGGING_CONFIG
-from utils import parse_datetime
+from utils import parse_datetime, dict_to_logfmt
 
 dictConfig(LOGGING_CONFIG)
 
@@ -53,13 +53,16 @@ def process_workflow_job():
     repository = job["repository"]["full_name"]
     action = job["action"]
 
+    context_details = {
+        "action": action,
+        "repository": repository,
+        "job_id": job_id,
+        "workflow": workflow,
+    }
+
     if action == "queued":
         # add to memory as timestamp
         jobs[job_id] = int(time_start.timestamp())
-        msg = (
-            f"action={action} repository={repository} job_id={job_id}"
-            f' workflow="{workflow}"'
-        )
 
     elif action == "in_progress":
         job_requested = jobs.get(job_id)
@@ -68,10 +71,7 @@ def process_workflow_job():
             time_to_start = 0
         else:
             time_to_start = (time_start - datetime.fromtimestamp(job_requested)).seconds
-        msg = (
-            f"action={action} repository={repository} job_id={job_id}"
-            f' workflow="{workflow}" time_to_start={time_to_start}'
-        )
+        context_details["time_to_start"] = time_to_start
 
     elif action == "completed":
         job_requested = jobs.get(job_id)
@@ -84,18 +84,15 @@ def process_workflow_job():
             ).seconds
             # delete from memory
             del jobs[job_id]
-        msg = (
-            f"action={action} repository={repository} job_id={job_id}"
-            f' workflow="{workflow}" time_to_finish={time_to_finish}'
-        )
+        context_details["time_to_finish"] = time_to_finish
     else:
         app.logger.warning(f"Unknown action {action}, removing from memory")
         if job_id in jobs:
             del jobs[job_id]
-        msg = None
+        context_details = None
 
-    if msg:
-        app.logger.info(msg)
+    if context_details:
+        app.logger.info(dict_to_logfmt(context_details))
     return True
 
 

src/utils.py

@@ -5,3 +5,21 @@ def parse_datetime(date: str) -> datetime:
     """Parse GitHub date to object"""
     exp = "%Y-%m-%dT%H:%M:%SZ"
     return datetime.strptime(date, exp)
+
+
+def dict_to_logfmt(data: dict) -> str:
+    """Convert a dict to logfmt string"""
+    outstr = list()
+    for k, v in data.items():
+        if v is None:
+            outstr.append(f"{k}=")
+            continue
+        if isinstance(v, bool):
+            v = "true" if v else "false"
+        elif isinstance(v, (dict, object, int)):
+            v = str(v)
+
+        if " " in v:
+            v = '"%s"' % v.replace('"', '\\"')
+        outstr.append(f"{k}={v}")
+    return " ".join(outstr)