Create ProviderETLHandler Abstract class for Workflows

apiserver/dora/service/workflows/sync/etl_github_actions_handler.py

@@ -4,6 +4,7 @@
 import pytz
 
 from dora.exapi.github import GithubApiService
+from dora.service.workflows.sync.etl_provider_handler import ProviderETLHandler
 from dora.store.models import UserIdentityProvider
 from dora.store.models.code import (
     RepoWorkflowProviders,
@@ -20,7 +21,7 @@
 WORKFLOW_PROCESSING_CHUNK_SIZE = 100
 
 
-class GithubActionsETLHandler:
+class GithubActionsETLHandler(ProviderETLHandler):
     def __init__(self, org_id: str, github_api_service: GithubApiService):
         self.org_id = org_id
         self._api: GithubApiService = github_api_service
@@ -72,11 +73,15 @@ def get_workflow_runs(
             )
             return []
 
-        bookmark.bookmark = self._get_new_bookmark_time_stamp(github_workflow_runs).isoformat()
+        bookmark.bookmark = self._get_new_bookmark_time_stamp(
+            github_workflow_runs
+        ).isoformat()
 
         return self._get_db_workflows(github_workflow_runs, str(repo_workflow.id))
 
-    def _get_new_bookmark_time_stamp(self, github_workflow_runs: List[Dict]) -> datetime:
+    def _get_new_bookmark_time_stamp(
+        self, github_workflow_runs: List[Dict]
+    ) -> datetime:
         """
         This method returns the new bookmark timestamp for the workflow runs.
         It returns the minimum timestamp of the pending jobs if there are any pending jobs.

apiserver/dora/service/workflows/sync/etl_provider_handler.py

@@ -0,0 +1,36 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+from dora.store.models.code import (
+    OrgRepo,
+    RepoWorkflow,
+    RepoWorkflowRunsBookmark,
+    RepoWorkflowRuns,
+)
+
+
+class ProviderETLHandler(ABC):
+    @abstractmethod
+    def check_pat_validity(self) -> bool:
+        """
+        This method checks if the PAT is valid.
+        :return: PAT details
+        :raises: Exception if PAT is invalid
+        """
+        pass
+
+    @abstractmethod
+    def get_workflow_runs(
+        self,
+        org_repo: OrgRepo,
+        repo_workflow: RepoWorkflow,
+        bookmark: RepoWorkflowRunsBookmark,
+    ) -> List[RepoWorkflowRuns]:
+        """
+        This method returns all workflow runs of a repo's workflow. After the bookmark date.
+        :param org_repo: OrgRepo object to get workflow runs for
+        :param repo_workflow: RepoWorkflow object to get workflow runs for
+        :param bookmark: Bookmark object to get all workflow runs after this date
+        :return: List of RepoWorkflowRuns objects
+        """
+        pass