Small improvements in app

Author: arponpes

app.py

@@ -1,79 +1,110 @@
-from flask import Flask, request, abort
+from datetime import datetime, timedelta
+from logging.config import dictConfig
+
+from flask import Flask, abort, request
 
-import logging
 from const import GithubHeaders
-from datetime import datetime, timedelta
 
-_LOGGER = logging.getLogger(__name__)
+dictConfig(
+    {
+        "version": 1,
+        "formatters": {
+            "default": {
+                "format": "[%(asctime)s]: %(levelname)s| %(message)s",
+            }
+        },
+        "handlers": {
+            "wsgi": {
+                "class": "logging.StreamHandler",
+                "stream": "ext://flask.logging.wsgi_errors_stream",
+                "formatter": "default",
+            }
+        },
+        "root": {"level": "INFO", "handlers": ["wsgi"]},
+    }
+)
 
 app = Flask(__name__)
 
 jobs = dict()
 
+
 def parse_datetime(date: str) -> datetime:
     exp = "%Y-%m-%dT%H:%M:%SZ"
     return datetime.strptime(date, exp)
 
+
+def prune_jobs():
+    # if the first job is older than 2 days, prune all jobs older than 2 days
+    # this might happen due to not completed jobs
+    if datetime.fromtimestamp(list(jobs.values())[0]) > datetime.now() - timedelta(
+        days=2
+    ):
+        app.logger.info("Pruning jobs")
+        for job_id, timestamp in jobs.items():
+            if (datetime.now() - datetime.fromtimestamp(timestamp)).days >= 2:
+                del jobs[job_id]
+
+
 def validate_origin_github() -> bool:
     userAgent = request.headers.get("User-Agent")
     if not userAgent.startswith("GitHub-Hookshot"):
-        _LOGGER.warning("User-Agent is {userAgent}")
+        app.logger.warning("User-Agent is {userAgent}")
         return False
 
     if request.headers.get("Content-Type") != "application/json":
-        _LOGGER.warning("Content is not JSON")
+        app.logger.warning("Content is not JSON")
         return False
 
     if not request.headers.get(GithubHeaders.EVENT.value):
-        _LOGGER.warning("No GitHub Event received!")
+        app.logger.warning("No GitHub Event received!")
         return False
 
     return True
 
+
 def process_workflow_job():
     job = request.get_json()
 
     job_id = job["workflow_job"]["run_id"]
-    name = job["workflow_job"]["workflow_name"]
+    workflow = job["workflow_job"]["workflow_name"]
     time_start = parse_datetime(job["workflow_job"]["started_at"])
     repository = job["repository"]["full_name"]
     action = job["action"]
-    NOW = datetime.now()
 
     if action == "queued":
         # add to memory as timestamp
         jobs[job_id] = int(time_start.timestamp())
-        msg = f"{NOW} {action=} {repository=} workflow={name} {job_id=} {time_start=}"
+        msg = f"{action=} {repository=} {workflow=} {job_id=}"
+        prune_jobs()
 
     elif action == "in_progress":
-        job_requested = jobs.get(job_id, None)
+        job_requested = jobs.get(job_id)
         if not job_requested:
-            _LOGGER.warning(f"Job {job_id} is {action} but not stored!")
+            app.logger.warning(f"Job {job_id} is {action} but not stored!")
             time_to_start = 0
         else:
-            time_to_start = (
-                time_start - datetime.fromtimestamp(job_requested)
-            ).seconds
-        msg = f"{NOW} {action=} {repository=} workflow={name} {job_id=} {time_to_start=}"
+            time_to_start = (time_start - datetime.fromtimestamp(job_requested)).seconds
+        msg = f"{action=} {repository=} {workflow=} {job_id=} {time_to_start=}"
 
     elif action == "completed":
-        job_requested = jobs.get(job_id, None)
+        job_requested = jobs.get(job_id)
         if not job_requested:
-            _LOGGER.warning(f"Job {job_id} is {action} but not stored!")
+            app.logger.warning(f"Job {job_id} is {action} but not stored!")
             time_to_finish = 0
         else:
             time_to_finish = (
-                time_start - datetime.fromtimestamp(job_requested)
+                parse_datetime(job["workflow_job"]["completed_at"]) - time_start
             ).seconds
             # delete from memory
             del jobs[job_id]
-        
-        msg = f"{NOW} {action=} {repository=} workflow={name} {job_id=} {time_to_finish=}"
 
-    print(msg)
+        msg = f"{action=} {repository=} {workflow=} {job_id=} {time_to_finish=}"
 
+    app.logger.info(msg)
     return True
 
+
 @app.route("/github-webhook", methods=["POST"])
 def github_webhook_process():
     if not validate_origin_github():
@@ -82,16 +113,14 @@ def github_webhook_process():
     event = request.headers.get(GithubHeaders.EVENT.value)
     command = f"process_{event}"
 
-    #if hasattr(command) and callable(getattr(command)):
     if command == "process_workflow_job":
-        _LOGGER.debug(f"Calling function {command}")
-        #response = getattr(command)()
+        app.logger.debug(f"Calling function {command}")
         response = process_workflow_job()
 
         if not response:
-            _LOGGER.error(f"Error calling {event} function")
+            app.logger.error(f"Error calling {event} function")
             return abort(500)
         return "OK"
 
-    _LOGGER.error(f"Unknown event type {event}, can't handle")
+    app.logger.error(f"Unknown event type {event}, can't handle")
     return abort(405)

const.py

@@ -1,6 +1,7 @@
 # StrEnum only in Python 3.11
 from enum import Enum
 
+
 class GithubHeaders(str, Enum):
     # NOTE: Flask manipulates as Capital-Word-Per-Section
     EVENT = "X-Github-Event"