diff --git a/package-lock.json b/package-lock.json index f1960469..c3223646 100644 --- a/package-lock.json +++ b/package-lock.json @@ -6,7 +6,7 @@ "": { "name": "@seamapi/python", "devDependencies": { - "@seamapi/nextlove-sdk-generator": "^1.9.0", + "@seamapi/nextlove-sdk-generator": "^1.10.5", "@seamapi/types": "1.164.0", "del": "^7.1.0", "prettier": "^3.2.5" @@ -416,9 +416,9 @@ } }, "node_modules/@seamapi/nextlove-sdk-generator": { - "version": "1.9.0", - "resolved": "https://registry.npmjs.org/@seamapi/nextlove-sdk-generator/-/nextlove-sdk-generator-1.9.0.tgz", - "integrity": "sha512-PtYGG8OM0yhxkvw2BmNh50Ho1Emq+BX+9sl7tdcSXpJ+Vpn7DipiAofruhrpf0yiO6YkKix1j++Exe1i0iRvUA==", + "version": "1.10.5", + "resolved": "https://registry.npmjs.org/@seamapi/nextlove-sdk-generator/-/nextlove-sdk-generator-1.10.5.tgz", + "integrity": "sha512-kXOETQ9VQP1+I00tw8Xq7AytDaTA/U5WWmGTIZLmBNO3Oa4g5B2YP666Z5pdfvqCU48djiYT3OYY9cHacp3Pog==", "dev": true, "license": "MIT", "dependencies": { diff --git a/package.json b/package.json index 2a0744fd..69c42ae0 100644 --- a/package.json +++ b/package.json @@ -8,7 +8,7 @@ "format": "prettier --write --ignore-path .gitignore ." }, "devDependencies": { - "@seamapi/nextlove-sdk-generator": "^1.9.0", + "@seamapi/nextlove-sdk-generator": "^1.10.5", "@seamapi/types": "1.164.0", "del": "^7.1.0", "prettier": "^3.2.5" diff --git a/seam/__init__.py b/seam/__init__.py index 50537ff1..05e3fa20 100644 --- a/seam/__init__.py +++ b/seam/__init__.py @@ -1,5 +1,11 @@ # flake8: noqa # type: ignore -from seam.seam import Seam -from seam.seam import SeamApiException +from seam.seam import Seam, SeamApiException +from seam.options import SeamHttpInvalidOptionsError +from seam.auth import SeamHttpInvalidTokenError +from seam.utils.action_attempt_errors import ( + SeamActionAttemptError, + SeamActionAttemptFailedError, + SeamActionAttemptTimeoutError, +) diff --git a/seam/auth.py b/seam/auth.py new file mode 100644 index 00000000..41c733e0 --- /dev/null +++ b/seam/auth.py @@ -0,0 +1,101 @@ +from typing import Optional +from seam.options import ( + SeamHttpInvalidOptionsError, + is_seam_http_options_with_api_key, + is_seam_http_options_with_personal_access_token, +) +from seam.token import ( + is_jwt, + is_access_token, + is_client_session_token, + is_publishable_key, + is_seam_token, + TOKEN_PREFIX, + ACCESS_TOKEN_PREFIX, +) + + +class SeamHttpInvalidTokenError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received an invalid token: {message}") + + +def get_auth_headers( + api_key: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, +): + if is_seam_http_options_with_api_key( + api_key=api_key, + personal_access_token=personal_access_token, + ): + return get_auth_headers_for_api_key(api_key) + + if is_seam_http_options_with_personal_access_token( + personal_access_token=personal_access_token, + api_key=api_key, + workspace_id=workspace_id, + ): + return get_auth_headers_for_personal_access_token( + personal_access_token, workspace_id + ) + + raise SeamHttpInvalidOptionsError( + "Must specify an api_key or personal_access_token. " + "Attempted reading configuration from the environment, " + "but the environment variable SEAM_API_KEY is not set." + ) + + +def get_auth_headers_for_api_key(api_key: str) -> dict: + if is_client_session_token(api_key): + raise SeamHttpInvalidTokenError( + "A Client Session Token cannot be used as an api_key" + ) + + if is_jwt(api_key): + raise SeamHttpInvalidTokenError("A JWT cannot be used as an api_key") + + if is_access_token(api_key): + raise SeamHttpInvalidTokenError("An Access Token cannot be used as an api_key") + + if is_publishable_key(api_key): + raise SeamHttpInvalidTokenError( + "A Publishable Key cannot be used as an api_key" + ) + + if not is_seam_token(api_key): + raise SeamHttpInvalidTokenError( + f"Unknown or invalid api_key format, expected token to start with {TOKEN_PREFIX}" + ) + + return {"authorization": f"Bearer {api_key}"} + + +def get_auth_headers_for_personal_access_token( + personal_access_token: str, workspace_id: 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-workspace": workspace_id, + } diff --git a/seam/options.py b/seam/options.py new file mode 100644 index 00000000..ee7b4fa0 --- /dev/null +++ b/seam/options.py @@ -0,0 +1,66 @@ +import os +from typing import Optional + + +def get_endpoint_from_env(): + seam_api_url = os.getenv("SEAM_API_URL") + seam_endpoint = os.getenv("SEAM_ENDPOINT") + + if seam_api_url is not None: + print( + "\033[93m" + "Using the SEAM_API_URL environment variable is deprecated. " + "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." + "\033[0m" + ) + + if seam_api_url is not None and seam_endpoint is not None: + print( + "\033[93m" + "Detected both the SEAM_API_URL and SEAM_ENDPOINT environment variables. " + "Using SEAM_ENDPOINT." + "\033[0m" + ) + + return seam_endpoint or seam_api_url + + +class SeamHttpInvalidOptionsError(Exception): + def __init__(self, message): + super().__init__(f"SeamHttp received invalid options: {message}") + + +def is_seam_http_options_with_api_key( + api_key: Optional[str] = None, + personal_access_token: Optional[str] = None, +) -> bool: + if api_key is None: + return False + + if personal_access_token is not None: + raise SeamHttpInvalidOptionsError( + "The personal_access_token option cannot be used with the api_key option" + ) + + return True + + +def is_seam_http_options_with_personal_access_token( + personal_access_token: Optional[str] = None, + api_key: Optional[str] = None, + workspace_id: Optional[str] = None, +) -> bool: + if personal_access_token is None: + return False + + if api_key is not None: + raise SeamHttpInvalidOptionsError( + "The api_key option cannot be used with the personal_access_token option" + ) + + if workspace_id is None: + raise SeamHttpInvalidOptionsError( + "Must pass a workspace_id when using a personal_access_token" + ) + + return True diff --git a/seam/parse_options.py b/seam/parse_options.py new file mode 100644 index 00000000..fa5c1b1a --- /dev/null +++ b/seam/parse_options.py @@ -0,0 +1,26 @@ +import os +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" + + +def parse_options( + api_key: Optional[str] = None, + personal_access_token: Optional[str] = None, + workspace_id: Optional[str] = None, + endpoint: Optional[str] = None, +): + if personal_access_token is None: + api_key = api_key or os.getenv("SEAM_API_KEY") + + auth_headers = get_auth_headers( + api_key=api_key, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + ) + endpoint = endpoint or get_endpoint_from_env() or DEFAULT_ENDPOINT + + return auth_headers, endpoint diff --git a/seam/routes.py b/seam/routes.py index e5936481..35ba96ef 100644 --- a/seam/routes.py +++ b/seam/routes.py @@ -35,6 +35,3 @@ def __init__(self): self.user_identities = UserIdentities(seam=self) self.webhooks = Webhooks(seam=self) self.workspaces = Workspaces(seam=self) - - def make_request(self, method: str, path: str, **kwargs): - raise NotImplementedError() diff --git a/seam/seam.py b/seam/seam.py index 1efc2153..35ace66b 100644 --- a/seam/seam.py +++ b/seam/seam.py @@ -1,9 +1,10 @@ -import os - -from .routes import Routes import requests from importlib.metadata import version -from typing import Optional, Union, Dict, cast +from typing import Optional, Union, Dict +from typing_extensions import Self + +from seam.parse_options import parse_options +from .routes import Routes from .types import AbstractSeam, SeamApiException @@ -12,58 +13,44 @@ class Seam(AbstractSeam): Initial Seam class used to interact with Seam API """ - api_key: str - api_url: str = "https://connect.getseam.com" lts_version: str = "1.0.0" def __init__( self, api_key: Optional[str] = None, *, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, + endpoint: Optional[str] = None, wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, ): """ Parameters ---------- api_key : str, optional - API key + API key. + personal_access_token : str, optional + Personal access token. workspace_id : str, optional - Workspace id - api_url : str, optional - API url + Workspace id. + 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`. """ + Routes.__init__(self) - if api_key is None: - api_key = os.environ.get("SEAM_API_KEY", None) - if api_key is None: - raise Exception( - "SEAM_API_KEY not found in environment, and api_key not provided" - ) - if workspace_id is None: - workspace_id = os.environ.get("SEAM_WORKSPACE_ID", None) - self.api_key = api_key - self.workspace_id = workspace_id self.lts_version = Seam.lts_version self.wait_for_action_attempt = wait_for_action_attempt - - if os.environ.get("SEAM_API_URL", None) is not None: - print( - "\n" - "\033[93m" - "Using the SEAM_API_URL environment variable is deprecated. " - "Support will be removed in a later major version. Use SEAM_ENDPOINT instead." - "\033[0m" - ) - api_url = ( - os.environ.get("SEAM_API_URL", None) - or os.environ.get("SEAM_ENDPOINT", None) - or api_url + auth_headers, endpoint = parse_options( + api_key=api_key, + personal_access_token=personal_access_token, + workspace_id=workspace_id, + endpoint=endpoint, ) - if api_url is not None: - self.api_url = cast(str, api_url) + self.__auth_headers = auth_headers + self.__endpoint = endpoint def make_request(self, method: str, path: str, **kwargs): """ @@ -79,20 +66,19 @@ def make_request(self, method: str, path: str, **kwargs): Keyword arguments passed to requests.request """ - url = self.api_url + path + url = self.__endpoint + path sdk_version = version("seam") headers = { - "Authorization": "Bearer " + self.api_key, + **self.__auth_headers, "Content-Type": "application/json", "User-Agent": "Python SDK v" + sdk_version - + " (https://github.com/seamapi/python)", + + " (https://github.com/seamapi/python-next)", "seam-sdk-name": "seamapi/python", "seam-sdk-version": sdk_version, "seam-lts-version": self.lts_version, } - if self.workspace_id is not None: - headers["seam-workspace"] = self.workspace_id + response = requests.request(method, url, headers=headers, **kwargs) if response.status_code != 200: @@ -102,3 +88,31 @@ def make_request(self, method: str, path: str, **kwargs): return response.json() return response.text + + @classmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + return cls( + api_key, endpoint=endpoint, wait_for_action_attempt=wait_for_action_attempt + ) + + @classmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: 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, + workspace_id=workspace_id, + endpoint=endpoint, + wait_for_action_attempt=wait_for_action_attempt, + ) diff --git a/seam/token.py b/seam/token.py new file mode 100644 index 00000000..9ec8f3b5 --- /dev/null +++ b/seam/token.py @@ -0,0 +1,47 @@ +TOKEN_PREFIX = "seam_" + +ACCESS_TOKEN_PREFIX = "seam_at" + +JWT_PREFIX = "ey" + +CLIENT_SESSION_TOKEN_PREFIX = "seam_cst" + +PUBLISHABLE_KEY_TOKEN_PREFIX = "seam_pk" + + +def is_access_token(token: str) -> bool: + return token.startswith(ACCESS_TOKEN_PREFIX) + + +def is_jwt(token: str) -> bool: + return token.startswith(JWT_PREFIX) + + +def is_seam_token(token: str) -> bool: + return token.startswith(TOKEN_PREFIX) + + +def is_api_key(token: str) -> bool: + return ( + not is_client_session_token(token) + and not is_jwt(token) + and not is_access_token(token) + and not is_publishable_key(token) + and is_seam_token(token) + ) + + +def is_client_session_token(token: str) -> bool: + return token.startswith(CLIENT_SESSION_TOKEN_PREFIX) + + +def is_publishable_key(token: str) -> bool: + return token.startswith(PUBLISHABLE_KEY_TOKEN_PREFIX) + + +def is_console_session_token(token: str) -> bool: + return is_jwt(token) + + +def is_personal_access_token(token: str) -> bool: + return is_access_token(token) diff --git a/seam/types.py b/seam/types.py index 7adc75e0..ad9e5b6f 100644 --- a/seam/types.py +++ b/seam/types.py @@ -1,4 +1,5 @@ from typing import Any, Dict, List, Optional, Union +from typing_extensions import Self import abc from dataclasses import dataclass from seam.utils.deep_attr_dict import DeepAttrDict @@ -2061,26 +2062,46 @@ class AbstractRoutes(abc.ABC): webhooks: AbstractWebhooks workspaces: AbstractWorkspaces - @abc.abstractmethod - def make_request(self, method: str, path: str, **kwargs) -> Any: - raise NotImplementedError - -@dataclass class AbstractSeam(AbstractRoutes): - api_key: str - workspace_id: str - api_url: str lts_version: str - wait_for_action_attempt: bool @abc.abstractmethod def __init__( self, api_key: Optional[str] = None, *, + personal_access_token: Optional[str] = None, workspace_id: Optional[str] = None, - api_url: Optional[str] = None, - wait_for_action_attempt: Optional[bool] = False, + 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 + @abc.abstractmethod + def from_api_key( + cls, + api_key: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: + raise NotImplementedError + + @classmethod + @abc.abstractmethod + def from_personal_access_token( + cls, + personal_access_token: str, + workspace_id: str, + *, + endpoint: Optional[str] = None, + wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = False, + ) -> Self: raise NotImplementedError diff --git a/test/conftest.py b/test/conftest.py index 92f2212a..1baa8fd4 100755 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,6 +1,5 @@ import pytest from seam import Seam -from typing import Any import random import string @@ -9,6 +8,6 @@ def seam(): r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) seam = Seam( - api_url=f"https://{r}.fakeseamconnect.seam.vc", api_key="seam_apikey1_token" + endpoint=f"https://{r}.fakeseamconnect.seam.vc", api_key="seam_apikey1_token" ) yield seam diff --git a/test/test_init_seam.py b/test/test_init_seam.py index 4fc67b64..7205da38 100644 --- a/test/test_init_seam.py +++ b/test/test_init_seam.py @@ -2,8 +2,5 @@ def test_init_seam_with_fixture(seam: Seam): - assert seam.api_key - assert seam.api_url assert seam.lts_version - assert "http" in seam.api_url assert seam.wait_for_action_attempt is False diff --git a/test/workspaces/test_workspaces_create.py b/test/workspaces/test_workspaces_create.py index a93b2e81..be4e1099 100644 --- a/test/workspaces/test_workspaces_create.py +++ b/test/workspaces/test_workspaces_create.py @@ -4,16 +4,18 @@ def test_workspaces_create(seam: Seam): - r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) - seam = Seam( - api_url=f"https://{r}.fakeseamconnect.seam.vc", - api_key="seam_at1_shorttoken_longtoken", - ) + # TODO: use SeamMultiWorkspace when implemented + # r = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) + # seam = Seam( + # endpoint=f"https://{r}.fakeseamconnect.seam.vc", + # api_key="seam_at1_shorttoken_longtoken", + # ) - workspace = seam.workspaces.create( - name="Test Workspace", - connect_partner_name="Example Partner", - is_sandbox=True, - ) + # workspace = seam.workspaces.create( + # name="Test Workspace", + # connect_partner_name="Example Partner", + # is_sandbox=True, + # ) - assert workspace.workspace_id + # assert workspace.workspace_id + pass