Add API and codecs for PUT /api/cloud_tokens (#16)

This implements the upcoming PUT /api/cloud_tokens API.

Author: velovix

brainframe/api/bf_codecs/__init__.py

@@ -22,4 +22,5 @@
     LicenseInfo,
     DATE_FORMAT,
 )
+from .cloud_codecs import CloudTokens, CloudUserInfo
 from .sorting import SortOptions, Ordering

brainframe/api/bf_codecs/cloud_codecs.py

@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+
+from .base_codecs import Codec
+
+
+@dataclass
+class CloudTokens(Codec):
+    """OAuth2 tokens for a BrainFrame Cloud user"""
+
+    access_token: str
+    """Provides access to the API"""
+
+    refresh_token: str
+    """Used to refresh the access token when it expires"""
+
+    def to_dict(self) -> dict:
+        return dict(self.__dict__)
+
+    @staticmethod
+    def from_dict(d: dict) -> "CloudTokens":
+        return CloudTokens(
+            access_token=d["access_token"],
+            refresh_token=d["refresh_token"],
+        )
+
+
+@dataclass
+class CloudUserInfo(Codec):
+    """Information on a BrainFrame Cloud user"""
+
+    sub: str
+    """A unique identifier for the user"""
+
+    email: str
+    """The user's email address"""
+
+    def to_dict(self):
+        return dict(self.__dict__)
+
+    @staticmethod
+    def from_dict(d: dict) -> "CloudUserInfo":
+        return CloudUserInfo(
+            sub=d["sub"],
+            email=d["email"],
+        )

brainframe/api/bf_errors.py

@@ -303,6 +303,11 @@ class IncompatibleCapsuleError(BaseAPIError):
     """
 
 
+@_server_origin_error()
+class UnauthorizedTokensError(BaseAPIError):
+    """The provided tokens do not correctly authorize a BrainFrame Cloud user"""
+
+
 class ServerNotReadyError(BaseAPIError):
     """The client was able to communicate with the server, but the server had
     not completed startup or was in an invalid state"""

brainframe/api/stub.py

@@ -21,7 +21,8 @@ class BrainFrameAPI(stubs.AlertStubMixin,
                     stubs.EncodingStubMixIn,
                     stubs.PremisesStubMixin,
                     stubs.UserStubMixin,
-                    stubs.LicenseStubMixIn):
+                    stubs.LicenseStubMixIn,
+                    stubs.CloudTokensStubMixin):
     """Provides access to BrainFrame API endpoints."""
 
     def __init__(self, server_url=None, credentials: Tuple[str, str] = None):

brainframe/api/stubs/__init__.py

@@ -12,4 +12,5 @@
 from .premises import PremisesStubMixin
 from .users import UserStubMixin
 from .licenses import LicenseStubMixIn
+from .cloud_tokens import CloudTokensStubMixin
 from .base_stub import BaseStub

brainframe/api/stubs/base_stub.py

@@ -58,6 +58,24 @@ def _get_json(self, api_url, timeout, params=None) -> Tuple[Any, dict]:
             return json.loads(resp.content), resp.headers
         return None, resp.headers
 
+    def _put_codec(self, api_url, timeout, codec: bf_codecs.Codec):
+        """Send a PUT request to the given URL.
+
+        :param api_url: The path to append to the base_url
+        :param timeout: The timeout to use for this request
+        :param codec: A codec to convert to JSON and send
+        :return: The JSON response as a dict, or None if none was sent
+        """
+        codec_data = codec.to_json()
+        resp = self._put(api_url,
+                         timeout,
+                         data=codec_data,
+                         content_type="application/json")
+
+        if resp.content:
+            return json.loads(resp.content)
+        return None
+
     def _put_json(self, api_url, timeout, json_data) -> Any:
         """Send a PUT request to the given URL.
 

brainframe/api/stubs/cloud_tokens.py

@@ -0,0 +1,17 @@
+from brainframe.api.bf_codecs import CloudTokens, CloudUserInfo
+from .base_stub import BaseStub, DEFAULT_TIMEOUT
+
+
+class CloudTokensStubMixin(BaseStub):
+    def set_cloud_tokens(self, cloud_tokens: CloudTokens,
+                         timeout=DEFAULT_TIMEOUT) -> CloudUserInfo:
+        """Authorizes the server against BrainFrame Cloud using the provided tokens.
+
+        :param cloud_tokens: The tokens to use for authorization
+        :param timeout: the timeout to use for this request
+        :return: Info on the BrainFrame Cloud user that corresponds to these tokens
+        """
+        req = "/api/cloud_tokens"
+        user_info_dict = self._put_codec(req, timeout, cloud_tokens)
+
+        return CloudUserInfo.from_dict(user_info_dict)