Add amt website client

amt/clients/clients.py

@@ -0,0 +1,92 @@
+import logging
+from abc import ABC, abstractmethod
+from datetime import datetime, timezone
+
+import httpx
+from amt.schema.github import RepositoryContent
+
+logger = logging.getLogger(__name__)
+
+
+class Client(ABC):
+    """
+    Abstract class which is used to set up HTTP clients that retrieve instruments from the
+    instrument register.
+    """
+
+    @abstractmethod
+    def __init__(self, max_retries: int = 3, timeout: int = 5) -> None:
+        transport = httpx.HTTPTransport(retries=max_retries)
+        self.client = httpx.Client(timeout=timeout, transport=transport)
+
+    @abstractmethod
+    def get_content(self, url: str) -> bytes:
+        """
+        This method should implement getting the content of an instrument from given URL.
+        """
+        ...
+
+    @abstractmethod
+    def list_content(self, url: str = "") -> RepositoryContent:
+        """
+        This method should implement getting list of instruments from given URL.
+        """
+        ...
+
+    def _get(self, url: str) -> httpx.Response:
+        """
+        Private function that performs a GET request to given URL.
+        """
+        response = self.client.get(url)
+        response.raise_for_status()
+        return response
+
+
+def get_client(repo_type: str) -> Client:
+    match repo_type:
+        case "github_pages":
+            return GitHubPagesClient()
+        case "github":
+            return GitHubClient()
+        case _:
+            raise ValueError(f"unknown repository type: {repo_type}")
+
+
+class GitHubPagesClient(Client):
+    def __init__(self) -> None:
+        super().__init__()
+
+    def get_content(self, url: str) -> bytes:
+        return super()._get(url).content
+
+    def list_content(self, url: str = "https://minbzk.github.io/instrument-registry/index.json") -> RepositoryContent:
+        response = super()._get(url)
+        return RepositoryContent.model_validate(response.json()["entries"])
+
+
+class GitHubClient(Client):
+    def __init__(self) -> None:
+        super().__init__()
+        self.client.event_hooks["response"] = [self._check_rate_limit]
+        # TODO(Berry): add authentication headers with event_hooks
+
+    def get_content(self, url: str) -> bytes:
+        return super()._get(url).content
+
+    def list_content(
+        self,
+        url: str = "https://api.github.com/repos/MinBZK/instrument-registry/contents/instruments?ref=main",
+    ) -> RepositoryContent:
+        response = super()._get(url)
+        return RepositoryContent.model_validate(response.json())
+
+    def _check_rate_limit(self, response: httpx.Response) -> None:
+        if "x-ratelimit-remaining" in response.headers:
+            remaining = int(response.headers["X-RateLimit-Remaining"])
+            if remaining == 0:
+                reset_timestamp = int(response.headers["X-RateLimit-Reset"])
+                reset_time = datetime.fromtimestamp(reset_timestamp, timezone.utc)  # noqa: UP017
+                wait_seconds = (reset_time - datetime.now(timezone.utc)).total_seconds()  # noqa: UP017
+                logger.warning(
+                    f"Rate limit exceeded. We need to wait for {wait_seconds} seconds. (not implemented yet)"
+                )

amt/schema/github.py

@@ -1,20 +1,20 @@
-from pydantic import BaseModel, HttpUrl, RootModel
+from pydantic import BaseModel, Field, HttpUrl, RootModel
 
 
 class Links(BaseModel):
     self: HttpUrl
-    git: HttpUrl
-    html: HttpUrl
+    git: HttpUrl | None = Field(default=None)
+    html: HttpUrl | None = Field(default=None)
 
 
 class ContentItem(BaseModel):
     name: str
     path: str
-    sha: str
+    sha: str | None = Field(default=None)
     size: int
-    url: HttpUrl
-    html_url: HttpUrl
-    git_url: HttpUrl
+    url: HttpUrl | None = Field(default=None)
+    html_url: HttpUrl | None = Field(default=None)
+    git_url: HttpUrl | None = Field(default=None)
     download_url: HttpUrl
     type: str
     _links: Links

amt/services/instruments.py

@@ -3,7 +3,7 @@
 
 import yaml
 
-from amt.clients.github import GitHubClient
+from amt.clients.clients import get_client
 from amt.core.exceptions import InstrumentError
 from amt.schema.github import RepositoryContent
 from amt.schema.instrument import Instrument
@@ -12,11 +12,11 @@
 
 
 class InstrumentsService:
-    def __init__(self) -> None:
-        self.client = GitHubClient()
+    def __init__(self, repo_type: str = "github_pages") -> None:
+        self.client = get_client(repo_type)
 
-    def fetch_github_content_list(self, folder: str = "instruments") -> RepositoryContent:
-        response = self.client.list_content(folder=folder)
+    def fetch_github_content_list(self) -> RepositoryContent:
+        response = self.client.list_content()
         return RepositoryContent.model_validate(response)
 
     def fetch_github_content(self, url: str) -> Instrument: