Add Typing

.pre-commit-config.yaml

@@ -49,3 +49,9 @@ repos:
       rev: v3.0.1
       hooks:
           - id: docker-compose-check
+
+    - repo: https://github.com/pre-commit/mirrors-mypy
+      rev: v1.10.0
+      hooks:
+        - id: mypy
+          additional_dependencies: [types-all]

pyproject.toml

@@ -7,19 +7,31 @@ output-format = "grouped"
 [tool.ruff.format]
 exclude = [".git", ".venv", ".mypy_cache", ".tox", "__pycache__"]
 
+[tool.mypy]
+check_untyped_defs = true
+disallow_any_generics = false
+disallow_incomplete_defs = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+show_error_codes = true
+warn_unused_ignores = true
+
 [tool.poetry]
 name = "github-webhook-server"
 version = "0.0.0"
 description = "A webhook server to manage Github reposotories and pull requests."
-authors = ["Meni Yakove <myakove@gmail.com>", "Ruth Netser <ruth.netser@gmail.com>"]
+authors = [
+  "Meni Yakove <myakove@gmail.com>",
+  "Ruth Netser <ruth.netser@gmail.com>",
+]
 readme = "README.md"
 license = "Apache-2.0"
 homepage = "https://github.com/myakove/github-webhook-server"
 repository = "https://github.com/myakove/github-webhook-server"
-packages = [{include = "webhook_server_container"}]
+packages = [{ include = "webhook_server_container" }]
 classifiers = [
-    "Programming Language :: Python :: 3",
-    "Operating System :: OS Independent",
+  "Programming Language :: Python :: 3",
+  "Operating System :: OS Independent",
 ]
 
 
@@ -29,7 +41,7 @@ Download = "https://quay.io/repository/myakove/github-webhook-server"
 
 [tool.poetry.dependencies]
 python = "^3.8"
-poetry-dynamic-versioning = {extras = ["plugin"], version = "^1.0.0"}
+poetry-dynamic-versioning = { extras = ["plugin"], version = "^1.0.0" }
 pygithub = "^2.0.0"
 pyyaml = "^6.0"
 build = "^1.0.0"

webhook_server_container/app.py

@@ -1,46 +1,43 @@
 import os
+from typing import Any, Dict
+
 from fastapi import Request
 import requests
 import urllib3
 from simple_logger.logger import get_logger
 
 
-from webhook_server_container.libs.github_api import GitHubApi
+from webhook_server_container.libs.github_api import ProcessGithubWehook
 from webhook_server_container.utils.constants import FASTAPI_APP
 
-APP_ROOT_PATH = "/webhook_server"
-REPOSITORIES_APP_API = {}
-MISSING_APP_REPOSITORIES = []
+APP_ROOT_PATH: str = "/webhook_server"
 urllib3.disable_warnings()
 
 LOGGER = get_logger(name="app", filename=os.environ.get("WEBHOOK_SERVER_LOG_FILE"))
 
 
 @FASTAPI_APP.get(f"{APP_ROOT_PATH}/healthcheck")
-def healthcheck():
-    return {"status": requests.status_codes.codes.ok, "message": "Alive"}
+def healthcheck() -> Dict[str, Any]:
+    return {"status": requests.codes.ok, "message": "Alive"}
 
 
 @FASTAPI_APP.post(APP_ROOT_PATH)
-async def process_webhook(request: Request):
-    process_failed_msg = {"status": requests.status_codes.codes.server_error, "Message": "Process failed"}
+async def process_webhook(request: Request) -> Dict[str, Any]:
+    log_prefix = request.headers.get("X-GitHub-Delivery", "")
+    process_failed_msg = {
+        "status": requests.codes.server_error,
+        "message": "Process failed",
+        "log_prefix": log_prefix,
+    }
     try:
         hook_data = await request.json()
     except Exception as ex:
         LOGGER.error(f"Error get JSON from request: {ex}")
         return process_failed_msg
 
     try:
-        api = GitHubApi(hook_data=hook_data)
-    except Exception as ex:
-        LOGGER.error(f"Failed to initialized GitHubApi instance: {ex}")
-        return process_failed_msg
-
-    github_event = request.headers.get("X-GitHub-Event")
-    event_log = f"Event type: {github_event}. event ID: {request.headers.get('X-GitHub-Delivery')}"
-    try:
-        api.process_hook(data=github_event, event_log=event_log)
-        return {"status": requests.status_codes.codes.ok, "Message": "process success"}
+        ProcessGithubWehook(hook_data=hook_data, headers=request.headers)
+        return {"status": requests.codes.ok, "message": "process success", "log_prefix": log_prefix}
 
     except Exception as ex:
         LOGGER.error(f"Failed to process hook: {ex}")

webhook_server_container/libs/config.py

@@ -1,25 +1,23 @@
 import os
+from typing import Any, Dict
 
 import yaml
 
 
 class Config:
-    def __init__(self):
-        self.data_dir = os.environ.get("WEBHOOK_SERVER_DATA_DIR", "/webhook_server")
-        self.config_path = os.path.join(self.data_dir, "config.yaml")
+    def __init__(self) -> None:
+        self.data_dir: str = os.environ.get("WEBHOOK_SERVER_DATA_DIR", "/webhook_server")
+        self.config_path: str = os.path.join(self.data_dir, "config.yaml")
         self.exists()
 
-    def exists(self):
+    def exists(self) -> None:
         if not os.path.isfile(self.config_path):
             raise FileNotFoundError(f"Config file {self.config_path} not found")
 
     @property
-    def data(self):
-        return self.get_data_from_config()
-
-    def get_data_from_config(self):
+    def data(self) -> Dict[str, Any]:
         with open(self.config_path) as fd:
             return yaml.safe_load(fd)
 
-    def get_repository(self, repository_name):
-        return self.data["repositories"].get(repository_name)
+    def get_repository(self, repository_name: str) -> Dict[str, Any]:
+        return self.data["repositories"].get(repository_name, {})