feat: Implement SeamMultiWorkspace client (#44)

* Bump generator to 1.10.1

* ci: Generate code

* Fix tests

* Bump generator to 1.10.2

* ci: Generate code

* Remove unnecessary import

* Bump generator to 1.10.3 to fix circular imports with a separate file options.py

* ci: Generate code

* Add SeamMultiWorkspace

* Add constants.py

* Add get_endpoint helper fn

* Add RequestMixin to define common make_request method

* ci: Format code

* Disable generate workflow

* Add abstract classes, add helper for getting PAT, fix create ws test

* Enforce keyword only args in WorkspacesProxy methods

* Define WorkspacesProxy outside SeamMultiWorkspace

* Fix proxy class usage

* Enable generation workflow

---------

Co-authored-by: Seam Bot <devops@getseam.com>

Author: andrii-balitskyi

.gitignore

@@ -321,3 +321,7 @@ dist
 .yarn/build-state.yml
 .yarn/install-state.gz
 .pnp.*
+
+# yalc
+.yalc/
+yalc.lock

seam/__init__.py

@@ -1,7 +1,9 @@
 # flake8: noqa
 # type: ignore
 
-from seam.seam import Seam, SeamApiException
+from seam.seam import Seam
+from seam.types import SeamApiException
+from seam.seam_multi_workspace import SeamMultiWorkspace
 from seam.options import SeamHttpInvalidOptionsError
 from seam.auth import SeamHttpInvalidTokenError
 from seam.routes.action_attempts import (

seam/auth.py

@@ -99,3 +99,31 @@ def get_auth_headers_for_personal_access_token(
         "authorization": f"Bearer {personal_access_token}",
         "seam-workspace": workspace_id,
     }
+
+
+def get_auth_headers_for_multi_workspace_personal_access_token(
+    personal_access_token: str,
+) -> dict:
+    if is_jwt(personal_access_token):
+        raise SeamHttpInvalidTokenError(
+            "A JWT cannot be used as a personal_access_token"
+        )
+
+    if is_client_session_token(personal_access_token):
+        raise SeamHttpInvalidTokenError(
+            "A Client Session Token cannot be used as a personal_access_token"
+        )
+
+    if is_publishable_key(personal_access_token):
+        raise SeamHttpInvalidTokenError(
+            "A Publishable Key cannot be used as a personal_access_token"
+        )
+
+    if not is_access_token(personal_access_token):
+        raise SeamHttpInvalidTokenError(
+            f"Unknown or invalid personal_access_token format, expected token to start with {ACCESS_TOKEN_PREFIX}"
+        )
+
+    return {
+        "authorization": f"Bearer {personal_access_token}",
+    }

seam/constants.py

@@ -0,0 +1,3 @@
+LTS_VERSION = "1.0.0"
+
+DEFAULT_ENDPOINT = "https://connect.getseam.com"

seam/options.py

@@ -1,6 +1,12 @@
 import os
 from typing import Optional
 
+from seam.constants import DEFAULT_ENDPOINT
+
+
+def get_endpoint(endpoint: Optional[str] = None):
+    return endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT
+
 
 def get_endpoint_from_env():
     seam_api_url = os.getenv("SEAM_API_URL")

seam/parse_options.py

@@ -2,9 +2,7 @@
 from typing import Optional
 
 from seam.auth import get_auth_headers
-from seam.options import get_endpoint_from_env
-
-DEFAULT_ENDPOINT = "https://connect.getseam.com"
+from seam.options import get_endpoint
 
 
 def parse_options(
@@ -21,6 +19,6 @@ def parse_options(
         personal_access_token=personal_access_token,
         workspace_id=workspace_id,
     )
-    endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT
+    endpoint = get_endpoint(endpoint)
 
     return auth_headers, endpoint

seam/request.py

@@ -0,0 +1,47 @@
+import requests
+from importlib.metadata import version
+
+from seam.types import AbstractRequestMixin, SeamApiException
+
+
+class RequestMixin(AbstractRequestMixin):
+    def make_request(self, method: str, path: str, **kwargs):
+        """
+        Makes a request to the API
+
+        Parameters
+        ----------
+        method : str
+          Request method
+        path : str
+          Request path
+        **kwargs
+          Keyword arguments passed to requests.request
+
+        Raises
+        ------
+        SeamApiException: If the response status code is not successful.
+        """
+
+        url = self._endpoint + path
+        sdk_version = version("seam")
+        headers = {
+            **self._auth_headers,
+            "Content-Type": "application/json",
+            "User-Agent": "Python SDK v"
+            + sdk_version
+            + " (https://github.com/seamapi/python-next)",
+            "seam-sdk-name": "seamapi/python",
+            "seam-sdk-version": sdk_version,
+            "seam-lts-version": self.lts_version,
+        }
+
+        response = requests.request(method, url, headers=headers, **kwargs)
+
+        if response.status_code != 200:
+            raise SeamApiException(response)
+
+        if "application/json" in response.headers["content-type"]:
+            return response.json()
+
+        return response.text

seam/seam.py

@@ -1,19 +1,19 @@
-import requests
-from importlib.metadata import version
 from typing import Optional, Union, Dict
 from typing_extensions import Self
 
+from .constants import LTS_VERSION
 from .parse_options import parse_options
+from .request import RequestMixin
 from .routes.routes import Routes
-from .types import AbstractSeam, SeamApiException
+from .types import AbstractSeam
 
 
-class Seam(AbstractSeam):
+class Seam(AbstractSeam, RequestMixin):
     """
     Initial Seam class used to interact with Seam API
     """
 
-    lts_version: str = "1.0.0"
+    lts_version: str = LTS_VERSION
 
     def __init__(
         self,
@@ -49,45 +49,8 @@ def __init__(
             workspace_id=workspace_id,
             endpoint=endpoint,
         )
-        self.__auth_headers = auth_headers
-        self.__endpoint = endpoint
-
-    def make_request(self, method: str, path: str, **kwargs):
-        """
-        Makes a request to the API
-
-        Parameters
-        ----------
-        method : str
-          Request method
-        path : str
-          Request path
-        **kwargs
-          Keyword arguments passed to requests.request
-        """
-
-        url = self.__endpoint + path
-        sdk_version = version("seam")
-        headers = {
-            **self.__auth_headers,
-            "Content-Type": "application/json",
-            "User-Agent": "Python SDK v"
-            + sdk_version
-            + " (https://github.com/seamapi/python-next)",
-            "seam-sdk-name": "seamapi/python",
-            "seam-sdk-version": sdk_version,
-            "seam-lts-version": self.lts_version,
-        }
-
-        response = requests.request(method, url, headers=headers, **kwargs)
-
-        if response.status_code != 200:
-            raise SeamApiException(response)
-
-        if "application/json" in response.headers["content-type"]:
-            return response.json()
-
-        return response.text
+        self._auth_headers = auth_headers
+        self._endpoint = endpoint
 
     @classmethod
     def from_api_key(

seam/seam_multi_workspace.py

@@ -0,0 +1,72 @@
+from typing import Dict, Optional, Union
+from typing_extensions import Self
+
+from .auth import get_auth_headers_for_multi_workspace_personal_access_token
+from .constants import LTS_VERSION
+from .options import get_endpoint
+from .request import RequestMixin
+from .types import AbstractSeamMultiWorkspace
+from .routes.workspaces import Workspaces
+
+
+class WorkspacesProxy:
+    """Proxy to expose only the 'create' and 'list' methods of Workspaces."""
+
+    def __init__(self, workspaces):
+        self._workspaces = workspaces
+
+    def list(self, **kwargs):
+        return self._workspaces.list(**kwargs)
+
+    def create(self, **kwargs):
+        return self._workspaces.create(**kwargs)
+
+
+class SeamMultiWorkspace(AbstractSeamMultiWorkspace, RequestMixin):
+    """
+    Seam class used to interact with Seam API without being scoped to any specific workspace.
+    """
+
+    lts_version: str = LTS_VERSION
+
+    def __init__(
+        self,
+        personal_access_token: str,
+        *,
+        endpoint: Optional[str] = None,
+        wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
+    ):
+        """
+        Parameters
+        ----------
+        personal_access_token : str, optional
+          Personal access token.
+        endpoint : str, optional
+          The API endpoint to which the request should be sent.
+        wait_for_action_attempt : bool or dict, optional
+          Controls whether to wait for an action attempt to complete, either as a boolean or as a dictionary specifying `timeout` and `poll_interval`. Defaults to `False`.
+        """
+
+        self.lts_version = SeamMultiWorkspace.lts_version
+        self.wait_for_action_attempt = wait_for_action_attempt
+        self._auth_headers = get_auth_headers_for_multi_workspace_personal_access_token(
+            personal_access_token
+        )
+        self._endpoint = get_endpoint(endpoint)
+
+        self._workspaces = Workspaces(seam=self)
+        self.workspaces = WorkspacesProxy(self._workspaces)
+
+    @classmethod
+    def from_personal_access_token(
+        cls,
+        personal_access_token: str,
+        *,
+        endpoint: Optional[str] = None,
+        wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
+    ) -> Self:
+        return cls(
+            personal_access_token=personal_access_token,
+            endpoint=endpoint,
+            wait_for_action_attempt=wait_for_action_attempt,
+        )

seam/types.py

@@ -1,8 +1,8 @@
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 from typing_extensions import Self
 import abc
 
-from seam.routes.types import AbstractRoutes
+from seam.routes.types import AbstractRoutes, Workspace
 
 
 class SeamApiException(Exception):
@@ -23,8 +23,15 @@ def __init__(
         )
 
 
-class AbstractSeam(AbstractRoutes):
+class AbstractRequestMixin(abc.ABC):
+    @abc.abstractmethod
+    def make_request(self, method: str, path: str, **kwargs):
+        raise NotImplementedError
+
+
+class AbstractSeam(AbstractRoutes, AbstractRequestMixin):
     lts_version: str
+    wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
 
     @abc.abstractmethod
     def __init__(
@@ -36,11 +43,6 @@ def __init__(
         endpoint: Optional[str] = None,
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
     ):
-        self.wait_for_action_attempt = wait_for_action_attempt
-        self.lts_version = AbstractSeam.lts_version
-
-    @abc.abstractmethod
-    def make_request(self, method: str, path: str, **kwargs) -> Any:
         raise NotImplementedError
 
     @classmethod
@@ -65,3 +67,49 @@ def from_personal_access_token(
         wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
     ) -> Self:
         raise NotImplementedError
+
+
+class AbstractSeamMultiWorkspaceWorkspaces(abc.ABC):
+    @abc.abstractmethod
+    def create(
+        self,
+        *,
+        connect_partner_name: str,
+        name: str,
+        is_sandbox: Optional[bool] = None,
+        webview_logo_shape: Optional[str] = None,
+        webview_primary_button_color: Optional[str] = None,
+    ) -> Workspace:
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def list(
+        self,
+    ) -> List[Workspace]:
+        raise NotImplementedError()
+
+
+class AbstractSeamMultiWorkspace(AbstractRequestMixin):
+    lts_version: str
+    wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]]
+
+    @abc.abstractmethod
+    def __init__(
+        self,
+        personal_access_token: str,
+        *,
+        endpoint: Optional[str] = None,
+        wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
+    ):
+        raise NotImplementedError
+
+    @classmethod
+    @abc.abstractmethod
+    def from_personal_access_token(
+        cls,
+        personal_access_token: str,
+        *,
+        endpoint: Optional[str] = None,
+        wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False,
+    ) -> Self:
+        raise NotImplementedError