Added user token handlers

Author: kompotkot

bugout/__main__.py

@@ -29,12 +29,12 @@ def main() -> None:
 
     parser.add_argument(
         "--brood_url",
-        default="http://localhost:7474",
+        default="http://localhost",
         help="Brood url",
     )
     parser.add_argument(
         "--spire_url",
-        default="http://localhost:7475",
+        default="http://localhost",
         help="Spire url",
     )
 

bugout/app.py

@@ -8,9 +8,12 @@
 
 
 class Bugout:
-    def __init__(self, brood_api_url: str, spire_api_url: str) -> None:
+    def __init__(
+        self, brood_api_url: Optional[str] = None, spire_api_url: Optional[str] = None
+    ) -> None:
         self.brood_api_url = brood_api_url
         self.spire_api_url = spire_api_url
+
         self.user = User(self.brood_api_url)
         self.group = Group(self.brood_api_url)
         self.journal = Journal(self.spire_api_url)
@@ -19,12 +22,17 @@ def __init__(self, brood_api_url: str, spire_api_url: str) -> None:
     def brood_url(self):
         return self.brood_api_url
 
+    @property
+    def spire_url(self):
+        return self.spire_api_url
+
     def brood_ping(self) -> Dict[str, str]:
         return calls.ping(self.brood_api_url)
 
     def spire_ping(self) -> Dict[str, str]:
         return calls.ping(self.spire_api_url)
 
+    # User handlers
     def get_user(self, token: str) -> data.BugoutUser:
         return self.user.get_user(token)
 
@@ -39,8 +47,24 @@ def create_user(
             username, email, password, autogenerated_token=autogenerated_token
         )
 
+    def create_token(self, username: str, password: str) -> data.BugoutToken:
+        return self.user.create_token(username, password)
+
+    def revoke_token(self, token: str) -> data.BugoutToken:
+        return self.user.revoke_token(token)
+
+    def update_token(
+        self,
+        token: str,
+        token_type: Optional[str] = None,
+        token_note: Optional[str] = None,
+    ) -> data.BugoutToken:
+        return self.user.update_token(token, token_type, token_note)
+
+    # Group handlers
     def get_group(self, group_id: str, token: str) -> data.BugoutGroup:
         return self.group.get_group(group_id, token)
 
+    # Journal handlers
     def get_journal(self, journal_id: str, token: str) -> data.BugoutJournal:
         return self.journal.get_journal(journal_id, token)

bugout/calls.py

@@ -8,6 +8,12 @@
 logger = logging.getLogger(__name__)
 
 
+class InvalidUrlSpec(ValueError):
+    """
+    Raised when an invalid url is specified.
+    """
+
+
 class BugoutUnexpectedResponse(Exception):
     """
     Raised when Bugout server response is unexpected (e.g. unparseable).

bugout/data.py

@@ -8,9 +8,17 @@
 
 @unique
 class Method(Enum):
+    delete = "delete"
     get = "get"
     post = "post"
-    delete = "delete"
+    put = "put"
+
+
+@unique
+class TokenType(Enum):
+    bugout = "bugout"
+    slack = "slack"
+    github = "github"
 
 
 class BugoutUser(BaseModel):
@@ -37,3 +45,9 @@ class BugoutJournal(BaseModel):
     name: str
     created_at: datetime
     updated_at: datetime
+
+
+class BugoutToken(BaseModel):
+    id: uuid.UUID
+    token_type: Optional[str]
+    token_note: Optional[str]

bugout/group.py

@@ -1,7 +1,7 @@
 import logging
 from typing import Any, Dict, List, Optional, Tuple
 
-from .calls import make_request
+from .calls import make_request, InvalidUrlSpec
 from .data import BugoutGroup, Method
 
 logger = logging.getLogger(__name__)
@@ -18,7 +18,9 @@ class Group:
     Represent a group from Bugout.
     """
 
-    def __init__(self, url) -> None:
+    def __init__(self, url: Optional[str] = None) -> None:
+        if url is None:
+            raise InvalidUrlSpec("Invalid brood url specified")
         self.url = url
 
     def _call(self, method: Method, path: str, **kwargs):

bugout/journal.py

@@ -1,7 +1,7 @@
 import logging
 from typing import Any, Dict, List, Optional, Tuple
 
-from .calls import make_request
+from .calls import make_request, InvalidUrlSpec
 from .data import BugoutJournal, Method
 
 logger = logging.getLogger(__name__)
@@ -18,7 +18,9 @@ class Journal:
     Represent a journal from Bugout.
     """
 
-    def __init__(self, url) -> None:
+    def __init__(self, url: Optional[str] = None) -> None:
+        if url is None:
+            raise InvalidUrlSpec("Invalid spire url specified")
         self.url = url
 
     def _call(self, method: Method, path: str, **kwargs):

bugout/user.py

@@ -1,8 +1,8 @@
 import logging
 from typing import Any, Dict, List, Optional, Tuple
 
-from .calls import make_request
-from .data import BugoutUser, Method
+from .calls import make_request, InvalidUrlSpec
+from .data import BugoutUser, BugoutToken, Method, TokenType
 
 logger = logging.getLogger(__name__)
 
@@ -13,43 +13,55 @@ class UserNotFound(Exception):
     """
 
 
+class TokenInvalidParameters(ValueError):
+    """
+    Raised when operations are applied to a token but invalid parameters are provided with which to
+    specify that token.
+    """
+
+
 class User:
     """
     Represent a user from Bugout.
     """
 
-    def __init__(self, url) -> None:
+    def __init__(self, url: Optional[str] = None) -> None:
+        if url is None:
+            raise InvalidUrlSpec("Invalid brood url specified")
         self.url = url
 
     def _call(self, method: Method, path: str, **kwargs):
         url = f"{self.url.rstrip('/')}/{path.rstrip('/')}"
         result = make_request(method=method, url=url, **kwargs)
         return result
 
-    def get_user(self, token: str) -> BugoutUser:
+    def _user_model_converter(self, user_object: Dict[str, Any]) -> BugoutUser:
+        return BugoutUser(
+            id=user_object.get("user_id"),
+            username=user_object.get("username"),
+            email=user_object.get("email"),
+            normalized_email=user_object.get("normalized_email"),
+            verified=user_object.get("verified"),
+            autogenerated=user_object.get("autogenerated"),
+            created_at=user_object.get("created_at"),
+            updated_at=user_object.get("updated_at"),
+        )
+
+    def get_user(self, token: str) -> Dict[str, Any]:
         get_user_path = "user"
         headers = {
             "Authorization": f"Bearer {token}",
         }
         result = self._call(method=Method.get, path=get_user_path, headers=headers)
-        return BugoutUser(
-            id=result.get("user_id"),
-            username=result.get("username"),
-            email=result.get("email"),
-            normalized_email=result.get("normalized_email"),
-            verified=result.get("verified"),
-            autogenerated=result.get("autogenerated"),
-            created_at=result.get("created_at"),
-            updated_at=result.get("updated_at"),
-        )
+        return self._user_model_converter(result)
 
     def create_user(
         self,
         username: str,
         email: str,
         password: str,
-        autogenerated_token: str = None,
-    ):
+        autogenerated_token: Optional[str] = None,
+    ) -> Dict[str, Any]:
         create_user_path = "user"
         data = {
             "username": username,
@@ -64,13 +76,50 @@ def create_user(
         result = self._call(
             method=Method.post, path=create_user_path, headers=headers, data=data
         )
-        return BugoutUser(
-            id=result.get("user_id"),
-            username=result.get("username"),
-            email=result.get("email"),
-            normalized_email=result.get("normalized_email"),
-            verified=result.get("verified"),
-            autogenerated=result.get("autogenerated"),
-            created_at=result.get("created_at"),
-            updated_at=result.get("updated_at"),
+        return self._user_model_converter(result)
+
+    def create_token(self, username: str, password: str) -> BugoutToken:
+        create_token_path = "token"
+        data = {
+            "username": username,
+            "password": password,
+        }
+        result = self._call(method=Method.post, path=create_token_path, data=data)
+        return BugoutToken(
+            id=result.get("access_token"),
+            token_type=result.get("token_type"),
         )
+
+    def revoke_token(self, token: str) -> BugoutToken:
+        create_token_path = "token"
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+        result = self._call(
+            method=Method.delete, path=create_token_path, headers=headers
+        )
+        return BugoutToken(id=result)
+
+    def update_token(
+        self,
+        token: str,
+        token_type: Optional[str] = None,
+        token_note: Optional[str] = None,
+    ) -> BugoutToken:
+        create_token_path = "token"
+
+        data = {"access_token": token}
+        if token_type is None and token_note is None:
+            raise TokenInvalidParameters(
+                "In order to update token, at least one of token_type, or token_note must be specified"
+            )
+        if token_type not in TokenType.__members__:
+            raise TokenInvalidParameters("Incorrect token type provided")
+
+        if token_type is not None:
+            data.update({"token_type": token_type})
+        if token_note is not None:
+            data.update({"token_note": token_note})
+
+        result = self._call(method=Method.put, path=create_token_path, data=data)
+        return BugoutToken(id=result, token_type=token_type, token_note=token_note)