Use Dataclass For Workflow Run

Signed-off-by: Hassan Abouelela <[email protected]>

Author: HassanAbouelela

pydis_site/apps/api/github_utils.py

@@ -1,7 +1,8 @@
 """Utilities for working with the GitHub API."""
-
+import dataclasses
 import datetime
 import math
+import typing
 
 import httpx
 import jwt
@@ -50,6 +51,29 @@ class RunPendingError(ArtifactProcessingError):
     status = 202
 
 
+@dataclasses.dataclass(frozen=True)
+class WorkflowRun:
+    """
+    A workflow run from the GitHub API.
+
+    https://docs.github.com/en/rest/actions/workflow-runs#get-a-workflow-run
+    """
+
+    name: str
+    head_sha: str
+    created_at: str
+    status: str
+    conclusion: str
+    artifacts_url: str
+
+    @classmethod
+    def from_raw(cls, data: dict[str, typing.Any]):
+        """Create an instance using the raw data from the API, discarding unused fields."""
+        return cls(**{
+            key.name: data[key.name] for key in dataclasses.fields(cls)
+        })
+
+
 def generate_token() -> str:
     """
     Generate a JWT token to access the GitHub API.
@@ -121,12 +145,12 @@ def authorize(owner: str, repo: str) -> httpx.Client:
         raise e
 
 
-def check_run_status(run: dict) -> str:
+def check_run_status(run: WorkflowRun) -> str:
     """Check if the provided run has been completed, otherwise raise an exception."""
-    created_at = datetime.datetime.strptime(run["created_at"], ISO_FORMAT_STRING)
+    created_at = datetime.datetime.strptime(run.created_at, ISO_FORMAT_STRING)
     run_time = datetime.datetime.now() - created_at
 
-    if run["status"] != "completed":
+    if run.status != "completed":
         if run_time <= MAX_RUN_TIME:
             raise RunPendingError(
                 f"The requested run is still pending. It was created "
@@ -135,12 +159,12 @@ def check_run_status(run: dict) -> str:
         else:
             raise RunTimeoutError("The requested workflow was not ready in time.")
 
-    if run["conclusion"] != "success":
+    if run.conclusion != "success":
         # The action failed, or did not run
-        raise ActionFailedError(f"The requested workflow ended with: {run['conclusion']}")
+        raise ActionFailedError(f"The requested workflow ended with: {run.conclusion}")
 
     # The requested action is ready
-    return run["artifacts_url"]
+    return run.artifacts_url
 
 
 def get_artifact(owner: str, repo: str, sha: str, action_name: str, artifact_name: str) -> str:
@@ -155,7 +179,8 @@ def get_artifact(owner: str, repo: str, sha: str, action_name: str, artifact_nam
 
         # Filter the runs for the one associated with the given SHA
         for run in runs["workflow_runs"]:
-            if run["name"] == action_name and sha == run["head_sha"]:
+            run = WorkflowRun.from_raw(run)
+            if run.name == action_name and sha == run.head_sha:
                 break
         else:
             raise NotFoundError(

pydis_site/apps/api/tests/test_github_utils.py

@@ -1,4 +1,6 @@
+import dataclasses
 import datetime
+import typing
 import unittest
 from unittest import mock
 
@@ -42,45 +44,46 @@ def encode(payload: dict, _: str, algorithm: str, *args, **kwargs) -> str:
 class CheckRunTests(unittest.TestCase):
     """Tests the check_run_status utility."""
 
+    run_kwargs: typing.Mapping = {
+        "name": "run_name",
+        "head_sha": "sha",
+        "status": "completed",
+        "conclusion": "success",
+        "created_at": datetime.datetime.now().strftime(github_utils.ISO_FORMAT_STRING),
+        "artifacts_url": "url",
+    }
+
     def test_completed_run(self):
         """Test that an already completed run returns the correct URL."""
         final_url = "some_url_string_1234"
 
-        result = github_utils.check_run_status({
-            "status": "completed",
-            "conclusion": "success",
-            "created_at": datetime.datetime.now().strftime(github_utils.ISO_FORMAT_STRING),
-            "artifacts_url": final_url,
-        })
+        kwargs = dict(self.run_kwargs, artifacts_url=final_url)
+        result = github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))
         self.assertEqual(final_url, result)
 
     def test_pending_run(self):
         """Test that a pending run raises the proper exception."""
+        kwargs = dict(self.run_kwargs, status="pending")
         with self.assertRaises(github_utils.RunPendingError):
-            github_utils.check_run_status({
-                "status": "pending",
-                "created_at": datetime.datetime.now().strftime(github_utils.ISO_FORMAT_STRING),
-            })
+            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))
 
     def test_timeout_error(self):
         """Test that a timeout is declared after a certain duration."""
+        kwargs = dict(self.run_kwargs, status="pending")
         # Set the creation time to well before the MAX_RUN_TIME
         # to guarantee the right conclusion
-        created = (
+        kwargs["created_at"] = (
             datetime.datetime.now() - github_utils.MAX_RUN_TIME - datetime.timedelta(minutes=10)
         ).strftime(github_utils.ISO_FORMAT_STRING)
 
         with self.assertRaises(github_utils.RunTimeoutError):
-            github_utils.check_run_status({"status": "pending", "created_at": created})
+            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))
 
     def test_failed_run(self):
         """Test that a failed run raises the proper exception."""
+        kwargs = dict(self.run_kwargs, conclusion="failed")
         with self.assertRaises(github_utils.ActionFailedError):
-            github_utils.check_run_status({
-                "status": "completed",
-                "conclusion": "failed",
-                "created_at": datetime.datetime.now().strftime(github_utils.ISO_FORMAT_STRING),
-            })
+            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))
 
 
 def get_response_authorize(_: httpx.Client, request: httpx.Request, **__) -> httpx.Response:
@@ -172,11 +175,16 @@ def get_response_get_artifact(request: httpx.Request, **_) -> httpx.Response:
 
         if request.method == "GET":
             if path == "/repos/owner/repo/actions/runs":
+                run = github_utils.WorkflowRun(
+                    name="action_name",
+                    head_sha="action_sha",
+                    created_at=datetime.datetime.now().strftime(github_utils.ISO_FORMAT_STRING),
+                    status="completed",
+                    conclusion="success",
+                    artifacts_url="artifacts_url"
+                )
                 return httpx.Response(
-                    200, request=request, json={"workflow_runs": [{
-                        "name": "action_name",
-                        "head_sha": "action_sha"
-                    }]}
+                    200, request=request, json={"workflow_runs": [dataclasses.asdict(run)]}
                 )
             elif path == "/artifact_url":
                 return httpx.Response(