refactor: use explicit token providers

Author: Adam Weeks

.github/workflows/tests.yml

@@ -18,6 +18,6 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - run: python -m pip install --upgrade pip
-      - run: python -m pip install -e ".[aws]" pytest ruff
+      - run: python -m pip install -e ".[dev]"
       - run: ruff check .
       - run: python -m pytest -q

README.md

@@ -1,72 +1,87 @@
 # Webex BYODS Python SDK
 
 Reusable Python utilities for managing Webex Bring Your Own Data Source (BYODS)
-data sources. The SDK supports data source registration, retrieval, updates,
-schema discovery, JWT claim inspection, service-app authentication, and data
-source token extension.
+data sources. The SDK supports registration, retrieval, updates, schema
+discovery, JWT claim inspection, service-app authentication, and data source
+token extension.
 
-This repository is the source of truth for the SDK. The companion CLI project
-uses this package as a consumer.
+The SDK does not read credentials from files, environment variables, or cloud
+secret managers. The host application owns secret retrieval and provides an
+explicit token provider to the SDK.
 
 ## Install
 
-Until the package is published, install it directly from a local checkout:
+Until publication, install from a local checkout:
 
 ```bash
 python3 -m pip install -e .
 ```
 
-Install AWS Secrets Manager support when running in Lambda or supplying a
-secret name:
+## Authenticate
 
-```bash
-python3 -m pip install -e ".[aws]"
-```
-
-## Configure authentication
+For an already-issued service-app token, use a static provider:
 
-Copy the safe template and provide service-app credentials plus a personal
-token that can request service-app tokens:
+```python
+from webex_byods import StaticAccessTokenProvider, WebexDataSourceClient
 
-```bash
-cp token-config.json.template token-config.json
+client = WebexDataSourceClient(
+    token_provider=StaticAccessTokenProvider("service-app-access-token")
+)
 ```
 
-Do not commit `token-config.json`.
-
-## Use the SDK
+For automatic service-app token acquisition, the host application supplies
+credentials and a personal-token provider. It can obtain those values from its
+own configuration, AWS Secrets Manager, or another secret system.
 
 ```python
-from webex_byods import TokenManager, WebexDataSourceManager
-
-token_manager = TokenManager(config_path="token-config.json")
-client = WebexDataSourceManager(
-    token_manager.get_service_app_token(), token_manager=token_manager
+from webex_byods import (
+    ServiceAppCredentials,
+    StaticAccessTokenProvider,
+    WebexDataSourceClient,
+    WebexServiceAppTokenProvider,
 )
 
-data_sources = client.list_all_data_sources()
-schemas = client.get_data_source_schemas()
+credentials = ServiceAppCredentials(
+    app_id="service-app-id",
+    client_id="service-app-client-id",
+    client_secret="service-app-client-secret",
+    target_org_id="target-org-id",
+)
+service_token_provider = WebexServiceAppTokenProvider(
+    credentials,
+    StaticAccessTokenProvider("personal-access-token"),
+)
+client = WebexDataSourceClient(token_provider=service_token_provider)
 ```
 
-BYODS API operations preserve a dictionary result contract containing
-`success`, `data` or `error`, and `status_code`. If a data-source API request
-returns HTTP 401, the client obtains a fresh service-app token and retries the
-request once.
+`OAuthRefreshTokenProvider` is available when the host application manages an
+OAuth refresh token. Providers use `InMemoryTokenStore` by default; applications
+that need persistence can supply their own `TokenStore` implementation.
 
-To rotate a data source nonce and obtain a fresh JWS token:
+## Use the client
 
 ```python
-result = token_manager.extend_data_source_token(
+data_sources = client.list_all_data_sources()
+schemas = client.get_data_source_schemas()
+
+result = client.extend_data_source_token(
     data_source_id="data-source-id",
     token_lifetime_minutes=1440,
 )
 ```
 
+BYODS API operations return dictionaries containing `success`, `data` or
+`error`, and `status_code`. A 401 response causes one retry with
+`token_provider.get_access_token(force_refresh=True)`.
+
+`WebexDataSourceManager` remains an alias for `WebexDataSourceClient` during
+the transition. `TokenManager` is deprecated; use
+`WebexServiceAppTokenProvider` instead.
+
 ## Development
 
 ```bash
-python3 -m pip install -e ".[aws]"
-python3 -m pip install pytest ruff
+python3 -m pip install -e ".[dev]"
 python3 -m pytest -q
 ruff check .
 ```

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "webex-byods-sdk"
-version = "0.1.0"
+version = "0.2.0"
 description = "Reusable Webex Bring Your Own Data Source management utilities"
 readme = "README.md"
 requires-python = ">=3.8"
@@ -14,7 +14,6 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-aws = ["boto3>=1.28.0"]
 dev = ["pytest>=7", "ruff>=0.5"]
 
 [project.urls]

tests/test_auth.py

@@ -1,15 +1,14 @@
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, call, patch
 
 import jwt
 
-from webex_byods import TokenManager
+from webex_byods import StaticAccessTokenProvider, WebexDataSourceClient
 
 
-@patch("webex_byods.auth.uuid.uuid4", return_value="new-nonce")
-@patch("webex_byods.auth.requests.put")
-@patch("webex_byods.auth.requests.get")
+@patch("webex_byods.client.uuid.uuid4", return_value="new-nonce")
+@patch("webex_byods.client.requests.request")
 def test_extend_data_source_token_updates_nonce_and_preserves_claim_values(
-    get, put, uuid4
+    request, uuid4
 ):
     jws_token = jwt.encode(
         {
@@ -26,12 +25,12 @@ def test_extend_data_source_token_updates_nonce_and_preserves_claim_values(
         "url": "https://gateway.example.com/data",
         "status": "active",
     }
-    get.return_value = get_response
     put_response = Mock(status_code=200)
     put_response.json.return_value = {"tokenExpiryTime": "2026-07-10T00:00:00Z"}
-    put.return_value = put_response
-    manager = TokenManager(config_path="unused.json")
-    manager.get_service_app_token = Mock(return_value="service-token")
+    request.side_effect = [get_response, put_response]
+    manager = WebexDataSourceClient(
+        token_provider=StaticAccessTokenProvider("service-token")
+    )
 
     result = manager.extend_data_source_token("source-1", 60)
 
@@ -43,12 +42,15 @@ def test_extend_data_source_token_updates_nonce_and_preserves_claim_values(
         "token_expiry": "2026-07-10T00:00:00Z",
         "message": "Data source token extended successfully. New expiry: 2026-07-10T00:00:00Z",
     }
-    put.assert_called_once_with(
+    assert request.call_args_list[1] == call(
+        "PUT",
         "https://webexapis.com/v1/dataSources/source-1",
         headers={
             "Authorization": "Bearer service-token",
+            "Accept": "application/json",
             "Content-Type": "application/json",
         },
+        timeout=30,
         json={
             "audience": "audience-from-token",
             "nonce": "new-nonce",
@@ -62,7 +64,9 @@ def test_extend_data_source_token_updates_nonce_and_preserves_claim_values(
 
 
 def test_extend_data_source_token_rejects_lifetime_over_limit():
-    result = TokenManager(config_path="unused.json").extend_data_source_token(
+    result = WebexDataSourceClient(
+        token_provider=StaticAccessTokenProvider("service-token")
+    ).extend_data_source_token(
         "source-1", 1441
     )
 

tests/test_client.py

@@ -1,6 +1,6 @@
 from unittest.mock import Mock, call, patch
 
-from webex_byods import WebexDataSourceManager
+from webex_byods import StaticAccessTokenProvider, WebexDataSourceClient
 
 
 @patch("webex_byods.client.requests.request")
@@ -9,7 +9,9 @@ def test_list_data_sources_returns_existing_result_shape(request):
     response.json.return_value = {"items": [{"id": "source-1"}]}
     request.return_value = response
 
-    result = WebexDataSourceManager("token").list_all_data_sources()
+    result = WebexDataSourceClient(
+        token_provider=StaticAccessTokenProvider("token")
+    ).list_all_data_sources()
 
     assert result == {
         "success": True,
@@ -35,12 +37,15 @@ def test_request_retries_once_after_token_refresh(request):
     refreshed_response.json.return_value = {"items": []}
     request.side_effect = [expired_response, refreshed_response]
     token_manager = Mock()
-    token_manager.get_service_app_token.return_value = "new-token"
+    token_manager.get_access_token.side_effect = ["old-token", "new-token"]
 
-    result = WebexDataSourceManager("old-token", token_manager=token_manager).list_all_data_sources()
+    result = WebexDataSourceClient(token_provider=token_manager).list_all_data_sources()
 
     assert result["success"] is True
-    token_manager.get_service_app_token.assert_called_once_with(force_refresh=True)
+    assert token_manager.get_access_token.call_args_list == [
+        call(force_refresh=False),
+        call(force_refresh=True),
+    ]
     assert request.call_args_list == [
         call(
             "GET",
@@ -72,7 +77,7 @@ def test_schema_cache_is_loaded_once(request):
         "items": [{"id": "schema-1", "serviceType": "Contact Center"}]
     }
     request.return_value = response
-    manager = WebexDataSourceManager("token")
+    manager = WebexDataSourceClient(token_provider=StaticAccessTokenProvider("token"))
 
     assert manager.get_schema_display_name("schema-1") == "Contact Center (schema-1...)"
     assert manager.get_available_schemas() == [

tests/test_token_providers.py

@@ -0,0 +1,99 @@
+from unittest.mock import Mock, call, patch
+
+from webex_byods import (
+    OAuthRefreshTokenProvider,
+    ServiceAppCredentials,
+    StaticAccessTokenProvider,
+    WebexServiceAppTokenProvider,
+)
+
+
+@patch("webex_byods.auth.requests.post")
+def test_oauth_refresh_provider_caches_and_forces_access_tokens(post):
+    first_response = Mock()
+    first_response.status_code = 200
+    first_response.json.return_value = {"access_token": "first-personal-token"}
+    second_response = Mock()
+    second_response.status_code = 200
+    second_response.json.return_value = {"access_token": "second-personal-token"}
+    post.side_effect = [first_response, second_response]
+    provider = OAuthRefreshTokenProvider(
+        client_id="client-id",
+        client_secret="client-secret",
+        refresh_token="refresh-token",
+    )
+
+    assert provider.get_access_token() == "first-personal-token"
+    assert provider.get_access_token() == "first-personal-token"
+    assert provider.get_access_token(force_refresh=True) == "second-personal-token"
+    assert post.call_args_list == [
+        call(
+            "https://webexapis.com/v1/access_token",
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            data={
+                "grant_type": "refresh_token",
+                "client_id": "client-id",
+                "client_secret": "client-secret",
+                "refresh_token": "refresh-token",
+            },
+        ),
+        call(
+            "https://webexapis.com/v1/access_token",
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            data={
+                "grant_type": "refresh_token",
+                "client_id": "client-id",
+                "client_secret": "client-secret",
+                "refresh_token": "refresh-token",
+            },
+        ),
+    ]
+
+
+@patch("webex_byods.auth.requests.post")
+def test_service_app_provider_uses_explicit_credentials_and_caches_token(post):
+    response = Mock()
+    response.json.return_value = {"access_token": "service-token"}
+    post.return_value = response
+    credentials = ServiceAppCredentials(
+        app_id="app-id",
+        client_id="client-id",
+        client_secret="client-secret",
+        target_org_id="org-id",
+    )
+    provider = WebexServiceAppTokenProvider(
+        credentials, StaticAccessTokenProvider("personal-token")
+    )
+
+    assert provider.get_access_token() == "service-token"
+    assert provider.get_access_token() == "service-token"
+
+    post.assert_called_once_with(
+        "https://webexapis.com/v1/applications/app-id/token",
+        headers={
+            "Authorization": "Bearer personal-token",
+            "Content-Type": "application/json",
+        },
+        json={
+            "clientId": "client-id",
+            "clientSecret": "client-secret",
+            "targetOrgId": "org-id",
+        },
+    )
+
+
+@patch("webex_byods.auth.requests.post")
+def test_service_app_provider_forces_a_new_service_token(post):
+    first_response = Mock()
+    first_response.json.return_value = {"access_token": "first-token"}
+    second_response = Mock()
+    second_response.json.return_value = {"access_token": "second-token"}
+    post.side_effect = [first_response, second_response]
+    provider = WebexServiceAppTokenProvider(
+        ServiceAppCredentials("app-id", "client-id", "client-secret", "org-id"),
+        StaticAccessTokenProvider("personal-token"),
+    )
+
+    assert provider.get_access_token() == "first-token"
+    assert provider.get_access_token(force_refresh=True) == "second-token"
+    assert post.call_count == 2