feat: jobmanager token authentication

github-runner-manager/pyproject.toml

@@ -80,7 +80,8 @@ select = ["E", "W", "F", "C", "N", "R", "D", "H"]
 # Ignore D107 Missing docstring in __init__
 ignore = ["W503", "D107", "E203"]
 # D100, D101, D102, D103, D104: Ignore docstring style issues in tests
-per-file-ignores = ["tests/*:D100,D101,D102,D103,D104,D205,D212"]
+# DCO020, DCO030, DCO050: Ignore docstring argument,returns,raises sections in tests
+per-file-ignores = ["tests/*:D100,D101,D102,D103,D104,D205,D212, DCO020, DCO030, DCO050"]
 docstring-convention = "google"
 # Check for properly formatted copyright header in each file
 copyright-check = "True"

github-runner-manager/src/github_runner_manager/configuration/jobmanager.py

@@ -10,6 +10,8 @@ class JobManagerConfiguration(BaseModel):
 
     Attributes:
        url: Base url of the job manager API.
+      token: Token to authenticate with the job manager API.
     """
 
     url: HttpUrl
+    token: str

github-runner-manager/src/github_runner_manager/jobmanager_api.py

@@ -0,0 +1,184 @@
+#  Copyright 2025 Canonical Ltd.
+#  See LICENSE file for licensing details.
+
+"""Module containing logic to handle calls to the jobmanager api."""
+from enum import Enum
+
+import jobmanager_client
+from jobmanager_client.exceptions import ApiException, NotFoundException
+from pydantic import BaseModel
+from urllib3.exceptions import RequestError
+
+
+class JobManagerAPIError(Exception):
+    """Base exception for JobManager API errors."""
+
+
+class JobManagerAPINotFoundError(JobManagerAPIError):
+    """Exception raised when a runner is not found in the JobManager API."""
+
+
+class JobStatus(str, Enum):
+    """Status of a job on the JobManager.
+
+    Attributes:
+        IN_PROGRESS: Represents a job that is in progress.
+        PENDING: Represents a job that is pending.
+    """
+
+    IN_PROGRESS = "IN_PROGRESS"
+    PENDING = "PENDING"
+
+
+class Job(BaseModel):
+    """Represents a job on the JobManagerAPI.
+
+    Attributes:
+        status: The status of the job.
+    """
+
+    status: str | None
+
+
+class RunnerStatus(str, Enum):
+    """Status of a runner on the JobManager.
+
+    Attributes:
+        IN_PROGRESS: Represents a runner that is in progress.
+        PENDING: Represents a runner that is pending.
+    """
+
+    IN_PROGRESS = "IN_PROGRESS"
+    PENDING = "PENDING"
+
+
+class RunnerRegistration(BaseModel):
+    """Represents a runner registration response from the JobManagerAPI.
+
+    Attributes:
+        id: The ID of the registered runner.
+        token: The token for the registered runner.
+    """
+
+    id: int
+    token: str
+
+
+class RunnerHealth(BaseModel):
+    """Represents the health status of a runner on the JobManagerAPI.
+
+    Attributes:
+        status: The health status of the runner.
+        deletable: Indicates if the runner can be deleted.
+    """
+
+    status: str
+    deletable: bool
+
+
+class JobManagerAPI:
+    """Handles interactions with the JobManager API."""
+
+    # The job manager api uses an autogenerated api client that uses urllib3 connection pools
+    # that are not multiprocessing safe: https://github.com/urllib3/urllib3/issues/850
+    # Therefore, we create a new ApiClient for each request and close resources
+    # to avoid issues with multiprocessing in the application.
+
+    def __init__(self, token: str, url: str):
+        """Initialize the JobManagerAPI with a token and URL.
+
+        Args:
+            token: The authentication token for the JobManager API.
+            url: The base URL for the JobManager API.
+        """
+        self._token = token
+        self.url = url
+
+    def get_runner_health(self, runner_id: int) -> RunnerHealth:
+        """Fetch the health status of a runner by its ID from the JobManager API.
+
+        Args:
+            runner_id: The ID of the runner to fetch health status for.
+
+        Raises:
+            JobManagerAPINotFoundError: If the runner with the given ID is not found.
+            JobManagerAPIError: If there is an error fetching the runner health.
+
+        Returns:
+            RunnerHealth: The health status of the runner.
+        """
+        with self._create_api_client() as api_client:
+            runners_api = jobmanager_client.RunnersApi(api_client=api_client)
+            try:
+                response = runners_api.get_runner_health_v1_runners_runner_id_health_get(runner_id)
+            except NotFoundException as err:
+                raise JobManagerAPINotFoundError(
+                    f"Health for runner with ID {runner_id} not found in JobManager API."
+                ) from err
+            except (ApiException, RequestError, ValueError) as exc:
+                raise JobManagerAPIError(
+                    f"Error fetching runner health for ID {runner_id}: {exc}"
+                ) from exc
+            return RunnerHealth(status=response.status, deletable=response.deletable)
+
+    def register_runner(self, name: str, labels: list[str]) -> RunnerRegistration:
+        """Register a new runner with the JobManager API.
+
+        Args:
+            name: The name of the runner to register.
+            labels: A list of labels to associate with the runner.
+
+        Returns:
+            RunnerRegistration: The registration details of the runner, including ID and token.
+
+        Raises:
+            JobManagerAPIError: If there is an error registering the runner.
+        """
+        with self._create_api_client() as api_client:
+            runners_api = jobmanager_client.RunnersApi(api_client=api_client)
+            runner_register_request = jobmanager_client.RunnerCreate(name=name, labels=labels)
+
+            try:
+                response = runners_api.register_runner_v1_runners_register_post(
+                    runner_register_request
+                )
+            except (ApiException, RequestError, ValueError) as exc:
+                raise JobManagerAPIError(f"Error registering runner: {exc}") from exc
+            return RunnerRegistration(id=response.id, token=response.token)
+
+    def get_job(self, job_id: int) -> Job:
+        """Fetch a job by its ID from the JobManager API.
+
+        Args:
+            job_id: The ID of the job to fetch.
+
+        Returns:
+            Job: The job object containing its status.
+
+        Raises:
+            JobManagerAPINotFoundError: If the job with the given ID is not found.
+            JobManagerAPIError: If there is an error fetching the job.
+        """
+        with self._create_api_client() as api_client:
+            jobs_api = jobmanager_client.JobsApi(api_client=api_client)
+            try:
+                response = jobs_api.get_job_v1_jobs_job_id_get(job_id)
+            except NotFoundException as err:
+                raise JobManagerAPINotFoundError(
+                    f"Job with ID {job_id} not found in JobManager API."
+                ) from err
+            except (ApiException, RequestError, ValueError) as exc:
+                raise JobManagerAPIError(f"Error fetching job with ID {job_id}: {exc}") from exc
+            return Job(status=response.status)
+
+    def _create_api_client(self) -> jobmanager_client.ApiClient:
+        """Create a new API client for the JobManager API.
+
+        Returns:
+            jobmanager_client.ApiClient: A new API client configured with the JobManager
+            API URL and token.
+        """
+        config = jobmanager_client.Configuration(host=self.url)
+        api_client = jobmanager_client.ApiClient(configuration=config)
+        api_client.set_default_header("Authorization", f"Bearer {self._token}")
+        return api_client