Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,9 @@ A workspace ID must be provided when using this method and all requests will be

.. code-block:: python

# Set the `SEAM_PERSONAL_ACCESS_TOKEN` and `SEAM_WORKSPACE_ID` environment variables
seam = Seam()

# Pass as an option to the constructor
seam = Seam(
personal_access_token="your-personal-access-token",
Expand Down Expand Up @@ -386,6 +389,9 @@ Obtain one from the Seam Console.

from seam import SeamWithoutWorkspace

# Set the `SEAM_PERSONAL_ACCESS_TOKEN` environment variable
seam = SeamWithoutWorkspace()

# Pass as an option to the constructor
seam = SeamWithoutWorkspace(personal_access_token="your-personal-access-token")

Expand Down
71 changes: 67 additions & 4 deletions seam/parse_options.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
import os
from typing import Optional

from .auth import get_auth_headers
from .options import get_endpoint
from .auth import (
get_auth_headers,
get_auth_headers_for_without_workspace_personal_access_token,
)
from .options import SeamInvalidOptionsError, get_endpoint


def parse_options(
Expand All @@ -11,8 +14,11 @@ def parse_options(
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")
api_key = api_key or get_api_key_from_env(personal_access_token)
personal_access_token = personal_access_token or get_personal_access_token_from_env(
api_key
)
workspace_id = workspace_id or os.getenv("SEAM_WORKSPACE_ID")

auth_headers = get_auth_headers(
api_key=api_key,
Expand All @@ -22,3 +28,60 @@ def parse_options(
endpoint = get_endpoint(endpoint)

return auth_headers, endpoint


def parse_without_workspace_options(
personal_access_token: Optional[str] = None,
endpoint: Optional[str] = None,
):
personal_access_token = personal_access_token or os.getenv(
"SEAM_PERSONAL_ACCESS_TOKEN"
)

if personal_access_token is None:
raise SeamInvalidOptionsError(
"Must specify a personal_access_token. "
"Attempted reading configuration from the environment, "
"but the environment variable SEAM_PERSONAL_ACCESS_TOKEN is not set."
)

auth_headers = get_auth_headers_for_without_workspace_personal_access_token(
personal_access_token
)
endpoint = get_endpoint(endpoint)

return auth_headers, endpoint


def get_api_key_from_env(personal_access_token: Optional[str]) -> Optional[str]:
"""Read the api_key from the environment.

A personal access token passed as an option takes precedence over the
environment, so the environment is not consulted when one is given.
"""

if personal_access_token is not None:
return None

api_key = os.getenv("SEAM_API_KEY")

if api_key is not None and os.getenv("SEAM_PERSONAL_ACCESS_TOKEN") is not None:
raise SeamInvalidOptionsError(
"Both SEAM_API_KEY and SEAM_PERSONAL_ACCESS_TOKEN environment variables "
"are defined. Please use only one authentication method."
)

return api_key


def get_personal_access_token_from_env(api_key: Optional[str]) -> Optional[str]:
"""Read the personal_access_token from the environment.

An api_key, whether passed as an option or read from the environment, takes
precedence, so the environment is not consulted when one is set.
"""

if api_key is not None:
return None

return os.getenv("SEAM_PERSONAL_ACCESS_TOKEN")
10 changes: 7 additions & 3 deletions seam/seam.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,13 +51,17 @@ def __init__(
It supports two authentication methods: API key or personal access token.

:param api_key: The API key for authenticating with Seam. Mutually
exclusive with personal_access_token
exclusive with personal_access_token. Read from the SEAM_API_KEY
environment variable when omitted
:type api_key: Optional[str]
:param personal_access_token: A personal access token for
authenticating with Seam. Mutually exclusive with api_key
authenticating with Seam. Mutually exclusive with api_key. Read
from the SEAM_PERSONAL_ACCESS_TOKEN environment variable when
omitted
:type personal_access_token: Optional[str]
:param workspace_id: The ID of the workspace to interact with.
Required when using a personal access token
Required when using a personal access token. Read from the
SEAM_WORKSPACE_ID environment variable when omitted
:type workspace_id: Optional[str]
:param endpoint: The custom API endpoint URL. If not provided, the
default Seam API endpoint will be used
Expand Down
18 changes: 10 additions & 8 deletions seam/seam_without_workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,8 @@
from typing_extensions import Self
from urllib3.util import Retry

from .auth import get_auth_headers_for_without_workspace_personal_access_token
from .constants import DEFAULT_TIMEOUT, LTS_VERSION
from .options import get_endpoint
from .parse_options import parse_without_workspace_options
from .client import SeamHttpClient
from .models import AbstractSeamWithoutWorkspace
from .routes.workspaces import Workspaces
Expand Down Expand Up @@ -47,7 +46,7 @@ class SeamWithoutWorkspace(AbstractSeamWithoutWorkspace):

def __init__(
self,
personal_access_token: str,
personal_access_token: Optional[str] = None,
*,
endpoint: Optional[str] = None,
wait_for_action_attempt: Optional[Union[bool, Dict[str, float]]] = True,
Expand All @@ -62,8 +61,9 @@ def __init__(
and configuration options.

:param personal_access_token: A personal access token for
authenticating with Seam
:type personal_access_token: str
authenticating with Seam. Read from the
SEAM_PERSONAL_ACCESS_TOKEN environment variable when omitted
:type personal_access_token: Optional[str]
:param endpoint: The custom API endpoint URL. If not provided,
the default Seam API endpoint will be used
:type endpoint: Optional[str]
Expand All @@ -80,15 +80,17 @@ def __init__(
niquests Session, for control the other options do not cover
:type niquests_options: Optional[Dict[str, Any]]

:raises SeamInvalidOptionsError: If no personal_access_token is provided
and the SEAM_PERSONAL_ACCESS_TOKEN environment variable is not set
:raises SeamInvalidTokenError: If the provided personal access token format is invalid
"""

self.lts_version = SeamWithoutWorkspace.lts_version
self.wait_for_action_attempt = wait_for_action_attempt
auth_headers = get_auth_headers_for_without_workspace_personal_access_token(
personal_access_token
auth_headers, endpoint = parse_without_workspace_options(
personal_access_token=personal_access_token,
endpoint=endpoint,
)
endpoint = get_endpoint(endpoint)

self.client = SeamHttpClient(
base_url=endpoint,
Expand Down
89 changes: 87 additions & 2 deletions test/env_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,16 @@

import pytest

from seam import Seam
from seam import Seam, SeamWithoutWorkspace
from seam.options import SeamInvalidOptionsError

ENV_VARS = ("SEAM_API_KEY", "SEAM_ENDPOINT", "SEAM_API_URL")
ENV_VARS = (
"SEAM_API_KEY",
"SEAM_ENDPOINT",
"SEAM_API_URL",
"SEAM_PERSONAL_ACCESS_TOKEN",
"SEAM_WORKSPACE_ID",
)


def cleanup_env():
Expand Down Expand Up @@ -114,3 +120,82 @@ def test_seam_api_key_env_variable_is_ignored_with_personal_access_token(server)

assert device.workspace_id == seed["seed_workspace_1"]
assert device.device_id == seed["august_device_1"]


def test_seam_constructor_uses_personal_access_token_env_variables(server):
endpoint, seed = server
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = seed["seam_at1_token"]
os.environ["SEAM_WORKSPACE_ID"] = seed["seed_workspace_1"]

seam = Seam(endpoint=endpoint)
device = seam.devices.get(device_id=seed["august_device_1"])

assert device.workspace_id == seed["seed_workspace_1"]
assert device.device_id == seed["august_device_1"]


def test_seam_rejects_both_api_key_and_personal_access_token_env_variables():
os.environ["SEAM_API_KEY"] = "some-api-key"
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = "some-access-token"
os.environ["SEAM_WORKSPACE_ID"] = "some-workspace-id"

with pytest.raises(
SeamInvalidOptionsError,
match=r"Both SEAM_API_KEY and SEAM_PERSONAL_ACCESS_TOKEN environment variables",
):
Seam()


def test_seam_personal_access_token_option_overrides_env_variables(server):
endpoint, seed = server
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = "some-invalid-token"
os.environ["SEAM_WORKSPACE_ID"] = seed["seed_workspace_1"]

seam = Seam(personal_access_token=seed["seam_at1_token"], endpoint=endpoint)
device = seam.devices.get(device_id=seed["august_device_1"])

assert device.workspace_id == seed["seed_workspace_1"]
assert device.device_id == seed["august_device_1"]


def test_seam_workspace_id_option_overrides_env_variables(server):
endpoint, seed = server
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = seed["seam_at1_token"]
os.environ["SEAM_WORKSPACE_ID"] = "some-invalid-workspace"

seam = Seam(workspace_id=seed["seed_workspace_1"], endpoint=endpoint)
device = seam.devices.get(device_id=seed["august_device_1"])

assert device.workspace_id == seed["seed_workspace_1"]
assert device.device_id == seed["august_device_1"]


def test_seam_without_workspace_constructor_uses_personal_access_token_env_variable(
server,
):
endpoint, seed = server
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = seed["seam_at1_token"]

seam = SeamWithoutWorkspace(endpoint=endpoint)
workspaces = seam.workspaces.list()

assert len(workspaces) > 0


def test_seam_without_workspace_personal_access_token_option_overrides_env_variables(
server,
):
endpoint, seed = server
os.environ["SEAM_PERSONAL_ACCESS_TOKEN"] = "some-invalid-token"

seam = SeamWithoutWorkspace(
personal_access_token=seed["seam_at1_token"], endpoint=endpoint
)
workspaces = seam.workspaces.list()

assert len(workspaces) > 0


def test_seam_without_workspace_requires_personal_access_token_env_variable():
with pytest.raises(SeamInvalidOptionsError, match=r"SEAM_PERSONAL_ACCESS_TOKEN"):
SeamWithoutWorkspace()
Loading