Added humbug endpoints

Author: kompotkot

bugout/app.py

@@ -4,6 +4,7 @@
 from . import data
 from .calls import ping
 from .group import Group
+from .humbug import Humbug
 from .journal import Journal
 from .user import User
 from .settings import BUGOUT_BROOD_URL, BUGOUT_SPIRE_URL, REQUESTS_TIMEOUT
@@ -26,6 +27,7 @@ def __init__(
 
         self.user = User(self.brood_api_url)
         self.group = Group(self.brood_api_url)
+        self.humbug = Humbug(self.spire_api_url)
         self.journal = Journal(self.spire_api_url)
 
     @property
@@ -549,3 +551,13 @@ def search(
     ) -> data.BugoutSearchResults:
         self.journal.timeout = timeout
         return self.journal.search(token, journal_id, query, limit, offset, content)
+
+    # Humbug
+    def get_humbug_integrations(
+        self,
+        token: Union[str, uuid.UUID],
+        group_id: Optional[Union[str, uuid.UUID]] = None,
+        timeout: float = REQUESTS_TIMEOUT,
+    ) -> data.BugoutHumbugIntegrationsList:
+        self.humbug.timeout = timeout
+        return self.humbug.get_humbug_integrations(token=token, group_id=group_id)

bugout/data.py

@@ -171,3 +171,16 @@ class BugoutSearchResults(BaseModel):
     next_offset: Optional[int]
     max_score: float
     results: List[BugoutSearchResult]
+
+
+class BugoutHumbugIntegration(BaseModel):
+    id: uuid.UUID
+    group_id: uuid.UUID
+    journal_id: uuid.UUID
+    journal_name: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+
+
+class BugoutHumbugIntegrationsList(BaseModel):
+    integrations: List[BugoutHumbugIntegration] = Field(default_factory=list)

bugout/humbug.py

@@ -0,0 +1,42 @@
+from typing import Any, Dict, Optional, Union
+import uuid
+
+from .calls import make_request, InvalidUrlSpec
+from .data import Method, Role, BugoutHumbugIntegrationsList
+from .settings import REQUESTS_TIMEOUT
+
+
+class Humbug:
+    """
+    Represent a humbug from Bugout.
+    """
+
+    def __init__(
+        self, url: Optional[str] = None, timeout: float = REQUESTS_TIMEOUT
+    ) -> None:
+        if url is None:
+            raise InvalidUrlSpec("Invalid spire url specified")
+        self.url = url
+        self.timeout = timeout
+
+    def _call(self, method: Method, path: str, **kwargs):
+        url = f"{self.url.rstrip('/')}/{path.rstrip('/')}"
+        result = make_request(method=method, url=url, timeout=self.timeout, **kwargs)
+        return result
+
+    def get_humbug_integrations(
+        self,
+        token: Union[str, uuid.UUID],
+        group_id: Optional[Union[str, uuid.UUID]] = None,
+    ) -> BugoutHumbugIntegrationsList:
+        humbug_path = "humbug/integrations"
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+        query_params = {}
+        if group_id is not None:
+            query_params.update({"group_id": group_id})
+        result = self._call(
+            method=Method.get, path=humbug_path, params=query_params, headers=headers
+        )
+        return BugoutHumbugIntegrationsList(**result)