Merge pull request #21 from bugout-dev/humbug-endpoints

Humbug endpoints

Author: zomglings

bugout/__init__.py

@@ -7,7 +7,7 @@
 
 __email__ = "engineering@bugout.dev"
 __license__ = "MIT"
-__version__ = "0.1.7"
+__version__ = "0.1.8"
 
 __all__ = (
     "__author__",

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
@@ -185,13 +187,15 @@ def get_user_tokens(
         token: Union[str, uuid.UUID],
         active: Optional[bool] = None,
         token_type: Optional[Union[str, data.TokenType]] = None,
+        restricted: Optional[bool] = None,
         timeout: float = REQUESTS_TIMEOUT,
     ) -> data.BugoutUserTokens:
         self.user.timeout = timeout
         return self.user.get_user_tokens(
             token=token,
             active=active,
             token_type=data.TokenType(token_type) if token_type is not None else None,
+            restricted=restricted,
         )
 
     # Group handlers
@@ -344,10 +348,18 @@ def delete_journal_scopes(
 
     # Journal handlers
     def create_journal(
-        self, token: Union[str, uuid.UUID], name: str, timeout: float = REQUESTS_TIMEOUT
+        self,
+        token: Union[str, uuid.UUID],
+        name: str,
+        journal_type: Optional[Union[str, data.JournalTypes]] = None,
+        timeout: float = REQUESTS_TIMEOUT,
     ) -> data.BugoutJournal:
         self.journal.timeout = timeout
-        return self.journal.create_journal(token=token, name=name)
+        if journal_type is None:
+            journal_type = data.JournalTypes.DEFAULT
+        return self.journal.create_journal(
+            token=token, name=name, journal_type=data.JournalTypes(journal_type)
+        )
 
     def list_journals(
         self, token: Union[str, uuid.UUID], timeout: float = REQUESTS_TIMEOUT
@@ -547,3 +559,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

@@ -33,6 +33,11 @@ class HolderType(Enum):
     group = "group"
 
 
+class JournalTypes(Enum):
+    DEFAULT = "default"
+    HUMBUG = "humbug"
+
+
 class BugoutUser(BaseModel):
     id: uuid.UUID = Field(alias="user_id")
     username: str
@@ -171,3 +176,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)

bugout/journal.py

@@ -14,6 +14,7 @@
     BugoutSearchResults,
     HolderType,
     Method,
+    JournalTypes,
 )
 from .settings import REQUESTS_TIMEOUT
 
@@ -107,11 +108,14 @@ def delete_journal_scopes(
         return BugoutJournalScopeSpecs(**result)
 
     # Journal module
-    def create_journal(self, token: Union[str, uuid.UUID], name: str) -> BugoutJournal:
+    def create_journal(
+        self,
+        token: Union[str, uuid.UUID],
+        name: str,
+        journal_type: JournalTypes,
+    ) -> BugoutJournal:
         journal_path = "journals/"
-        json = {
-            "name": name,
-        }
+        json = {"name": name, "journal_type": journal_type.value}
         headers = {
             "Authorization": f"Bearer {token}",
         }

bugout/user.py

@@ -235,6 +235,7 @@ def get_user_tokens(
         token: Union[str, uuid.UUID],
         active: Optional[bool] = None,
         token_type: Optional[TokenType] = None,
+        restricted: Optional[bool] = None,
     ) -> BugoutUserTokens:
         get_user_tokens_path = "tokens"
         headers = {
@@ -245,6 +246,8 @@ def get_user_tokens(
             query_params.update({"active": str(int(active))})
         if token_type is not None:
             query_params.update({"token_type": token_type.value})
+        if restricted is not None:
+            query_params.update({"restricted": str(int(restricted))})
         result = self._call(
             method=Method.get,
             path=get_user_tokens_path,