Main skeleton

Added get user, group, journal and cli.

Author: kompotkot

.gitignore

@@ -127,3 +127,13 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Pycharm
+.idea
+
+# VS Code
+.vscode/
+
+# Custom
+.secrets/
+.bugout/

bugout/__init__.py

@@ -0,0 +1,17 @@
+"""
+Bugout Python API
+"""
+__author__ = "Bugout"
+__maintainer__ = __author__
+
+__email__ = "neeraj@bugout.dev"
+__license__ = "MIT"
+__version__ = "0.0.1"
+
+__all__ = (
+    "__author__",
+    "__email__",
+    "__license__",
+    "__maintainer__",
+    "__version__",
+)

bugout/__main__.py

@@ -0,0 +1,115 @@
+import argparse
+import logging
+import textwrap
+
+from .app import Bugout
+from .user import User
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def user_get_handler(args: argparse.Namespace) -> None:
+    """
+    Handler for "users get" subcommand.
+    """
+    bugout = Bugout(url=args.url, port=args.port)
+    user_obj = User(bugout)
+
+    result = user_obj.get_user(token=args.token)
+
+    print(result)
+
+
+def user_create_handler(args: argparse.Namespace) -> None:
+    """
+    Handler for "users create" subcommand.
+    """
+    bugout = Bugout(url=args.url, port=args.port)
+    user_obj = User(bugout)
+
+    result = user_obj.create_user(
+        username=args.username,
+        email=args.email,
+        password=args.password,
+        autogenerated_token=args.autogenerated_token,
+    )
+
+    print(result)
+
+
+def main() -> None:
+    bugout_description = textwrap.dedent(
+        """\
+        Bugout API: Tools for helping with Bugout API.
+        """
+    )
+    parser = argparse.ArgumentParser(prog="bugout", description=bugout_description)
+    parser.set_defaults(func=lambda _: parser.print_help())
+    subcommands = parser.add_subparsers(description="Journal commands")
+
+    parser.add_argument(
+        "--url",
+        default="localhost",
+        help="Dest url",
+    )
+    parser.add_argument(
+        "--port",
+        default="443",
+        help="Dest port",
+    )
+
+    # Users handler
+    parser_users = subcommands.add_parser(
+        "users", description="Work with Bugout users API handlers"
+    )
+    parser_users.set_defaults(
+        func=lambda _: parser_users.print_help(), subparser="users"
+    )
+    subcommands_users = parser_users.add_subparsers(description="Users commands")
+
+    parser_user_get = subcommands_users.add_parser("get", description="Get user")
+    parser_user_get.set_defaults(subcommand="get")
+    parser_user_get.add_argument(
+        "-t",
+        "--token",
+        required=True,
+        help="User token",
+    )
+    parser_user_get.set_defaults(func=user_get_handler)
+
+    parser_user_create = subcommands_users.add_parser(
+        "create", description="Create user"
+    )
+    parser_user_create.set_defaults(subcommand="create")
+    parser_user_create.add_argument(
+        "-u",
+        "--username",
+        required=True,
+        help="Username",
+    )
+    parser_user_create.add_argument(
+        "-e",
+        "--email",
+        required=True,
+        help="Email",
+    )
+    parser_user_create.add_argument(
+        "-p",
+        "--password",
+        required=True,
+        help="Password",
+    )
+    parser_user_create.add_argument(
+        "-a",
+        "--autogenerated_token",
+        help="Token",
+    )
+    parser_user_create.set_defaults(func=user_create_handler)
+
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

bugout/app.py

@@ -0,0 +1,15 @@
+from .data import Method
+from .calls import ApiCalls
+
+
+class Bugout:
+    def __init__(self, url, port) -> None:
+        self.url = url
+        self.port = port
+
+    def _call(self, method: Method, path: str, **kwargs):
+        url = f"{self.url.rstrip('/')}:{self.port}/{path.rstrip('/')}"
+
+        result = ApiCalls.make_request(method=method, url=url, **kwargs)
+
+        return result

bugout/calls.py

@@ -0,0 +1,34 @@
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+
+import requests
+
+from .data import Method
+
+logger = logging.getLogger(__name__)
+
+
+class BugoutUnexpectedResponse(Exception):
+    """
+    Raised when Bugout server response is unexpected (e.g. unparseable).
+    """
+
+
+class ApiCalls:
+    """
+    Handle REST API requests.
+    """
+
+    @staticmethod
+    def make_request(method: Method, url: str, **kwargs) -> Optional[Dict[str, Any]]:
+        response_body = None
+
+        try:
+            r = requests.request(method.value, url=url, **kwargs)
+            r.raise_for_status()
+            response_body = r.json()
+        except Exception as e:
+            logger.error(f"Exception {str(e)}")
+            raise
+
+        return response_body

bugout/data.py

@@ -0,0 +1,24 @@
+from enum import Enum, unique
+from typing import Any, Dict, List, Optional, Tuple
+import uuid
+
+from pydantic import BaseModel
+
+
+@unique
+class Method(Enum):
+    get = "get"
+    post = "post"
+    delete = "delete"
+
+
+class BroodUser(BaseModel):
+    id: uuid.UUID
+    username: Optional[str]
+    email: Optional[str]
+    token: Optional[uuid.UUID]
+
+
+class BroodGroup(BaseModel):
+    id: uuid.UUID
+    group_name: Optional[str]

bugout/group.py

@@ -0,0 +1,32 @@
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+
+from .app import Bugout
+from .data import Method
+
+logger = logging.getLogger(__name__)
+
+
+class GroupNotFound(Exception):
+    """
+    Raised on actions that involve group which are not present in the database.
+    """
+
+
+class Group(Bugout):
+    """
+    Represent a group from Bugout.
+    """
+
+    def __init__(self, bugout: Bugout) -> None:
+        super().__init__(bugout.url, bugout.port)
+
+    def get_group(self, group_id: str, token: str) -> Optional[Dict[str, Any]]:
+        get_group_path = f"groups/{group_id}"
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+
+        result = super()._call(method=Method.get, path=get_group_path, headers=headers)
+        return result

bugout/journals.py

@@ -0,0 +1,34 @@
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+
+from .app import Bugout
+from .data import Method
+
+logger = logging.getLogger(__name__)
+
+
+class JournalNotFound(Exception):
+    """
+    Raised on actions that involve journal which are not present in the database.
+    """
+
+
+class Journal(Bugout):
+    """
+    Represent a journal from Bugout.
+    """
+
+    def __init__(self, bugout: Bugout) -> None:
+        super().__init__(bugout.url, bugout.port)
+
+    def get_journal(self, journal_id: str, token: str) -> Optional[Dict[str, Any]]:
+        get_journal_path = f"journals/{journal_id}"
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+
+        result = super()._call(
+            method=Method.get, path=get_journal_path, headers=headers
+        )
+        return result

bugout/user.py

@@ -0,0 +1,59 @@
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+import uuid
+
+from .app import Bugout
+from .data import Method
+
+logger = logging.getLogger(__name__)
+
+
+class UserNotFound(Exception):
+    """
+    Raised on actions that involve user which are not present in the database.
+    """
+
+
+class User(Bugout):
+    """
+    Represent a user from Bugout.
+    """
+
+    def __init__(self, bugout: Bugout) -> None:
+        super().__init__(bugout.url, bugout.port)
+
+    def get_user(self, token: str) -> Optional[Dict[str, Any]]:
+        get_user_path = "user"
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+
+        result = super()._call(method=Method.get, path=get_user_path, headers=headers)
+        return result
+
+    def create_user(
+        self,
+        username: str,
+        email: str,
+        password: str,
+        autogenerated_token: Optional[str],
+    ):
+        create_user_path = "user"
+
+        data = {
+            "username": username,
+            "email": email,
+            "password": password,
+        }
+        headers = None
+
+        if autogenerated_token:
+            headers = {
+                "x-bugout-installation-token": autogenerated_token,
+            }
+
+        result = super()._call(
+            method=Method.post, path=create_user_path, headers=headers, data=data
+        )
+        return result

requirements.dev.txt

@@ -0,0 +1,10 @@
+appdirs==1.4.4
+black==20.8b1
+click==7.1.2
+mypy==0.790
+mypy-extensions==0.4.3
+pathspec==0.8.1
+regex==2020.11.13
+toml==0.10.2
+typed-ast==1.4.1
+typing-extensions==3.7.4.3