Update DynamoDB TokenStorage example + mypy config

docs/examples/token_storage/dynamodb_token_storage.py

@@ -7,7 +7,7 @@
 import boto3
 
 import globus_sdk
-from globus_sdk.token_storage import StorageAdapter
+from globus_sdk.token_storage import TokenStorage, TokenStorageData
 
 CLIENT_ID = "61338d24-54d5-408f-a10d-66c06b59f6d2"
 tablename = "example-globus-tokenstorage"
@@ -19,11 +19,13 @@
 )
 
 boto_client = boto3.client("dynamodb")
-auth_client = globus_sdk.NativeAppAuthClient(CLIENT_ID)
 
 
-class DynamoDBStorageAdapter(StorageAdapter):
-    def __init__(self, client, tablename: str, namespace: str = "DEFAULT") -> None:
+class DynamoDBTokenStorage(TokenStorage):
+
+    def __init__(
+        self, client: t.Any, tablename: str, namespace: str = "DEFAULT"
+    ) -> None:
         """
         :param client: A boto3 DyanmoDB client to use
         :param tablename: The name of the dynamodb table to use
@@ -53,25 +55,39 @@ def _compute_key(self, resource_server: str) -> str:
         """
         return f"{self.namespace}:{resource_server}"
 
-    def store(self, token_response: globus_sdk.OAuthTokenResponse) -> None:
-        for resource_server, token_data in token_response.by_resource_server.items():
+    def store_token_data_by_resource_server(
+        self, token_data_by_resource_server: t.Mapping[str, TokenStorageData]
+    ) -> None:
+        for resource_server, token_data in token_data_by_resource_server.items():
             key = self._compute_key(resource_server)
             dynamo_item = {
                 "token_data_id": {"S": key},
                 "resource_server": {"S": resource_server},
-                "access_token": {"S": token_data["access_token"]},
-                "refresh_token": {"S": token_data["refresh_token"]},
-                "expires_at_seconds": {"N": str(token_data["expires_at_seconds"])},
-                "scope": {"S": token_data["scope"]},
+                "access_token": {"S": token_data.access_token},
+                "refresh_token": {"S": token_data.refresh_token},
+                "expires_at_seconds": {"N": str(token_data.expires_at_seconds)},
+                "scope": {"S": token_data.scope},
             }
             # avoid setting `refresh_token` if it is null (meaning the
             # login flow used access tokens only)
-            if token_data["refresh_token"] is None:
+            if token_data.refresh_token is None:
                 del dynamo_item["refresh_token"]
 
             self.client.put_item(TableName=self.tablename, Item=dynamo_item)
 
-    def get_token_data(self, resource_server: str) -> dict[str, t.Any] | None:
+    def remove_token_data(self, resource_server: str) -> bool:
+        key = self._compute_key(resource_server)
+
+        deletion_result = self.client.delete_item(
+            TableName=self.tablename,
+            Key={"token_data_id": {"S": key}},
+            ReturnValues="ALL_OLD",
+        )
+
+        # Attributes are returned if a value was deleted, but not otherwise
+        return "Attributes" in deletion_result
+
+    def get_token_data(self, resource_server: str) -> TokenStorageData | None:
         key = self._compute_key(resource_server)
 
         wrapped_item = self.client.get_item(
@@ -83,13 +99,36 @@ def get_token_data(self, resource_server: str) -> dict[str, t.Any] | None:
             return None
 
         dynamo_item = wrapped_item["Item"]
-        return {
-            "resource_server": dynamo_item["resource_server"]["S"],
-            "access_token": dynamo_item["access_token"]["S"],
-            "refresh_token": dynamo_item.get("refresh_token", {"S": None})["S"],
-            "expires_at_seconds": int(dynamo_item["expires_at_seconds"]["N"]),
-            "scope": dynamo_item["scope"]["S"],
-        }
+        return TokenStorageData(
+            resource_server=dynamo_item["resource_server"]["S"],
+            identity_id=None,
+            token_type="Bearer",
+            scope=dynamo_item["scope"]["S"],
+            access_token=dynamo_item["access_token"]["S"],
+            refresh_token=dynamo_item.get("refresh_token", {"S": None})["S"],
+            expires_at_seconds=int(dynamo_item["expires_at_seconds"]["N"]),
+        )
+
+    def get_token_data_by_resource_server(self) -> dict[str, TokenStorageData]:
+        print("WARNING: scanning dynamodb tables is an expensive operation")
+        print("WARNING: consider whether or not you want to use this in production")
+
+        scan_result = self.client.scan(TableName=self.tablename, ConsistentRead=True)
+
+        by_resource_server: dict[str, TokenStorageData] = {}
+        for item in scan_result["Items"]:
+            resource_server = item["resource_server"]["S"]
+            by_resource_server[resource_server] = TokenStorageData(
+                resource_server=resource_server,
+                identity_id=None,
+                token_type="Bearer",
+                scope=item["scope"]["S"],
+                access_token=item["access_token"]["S"],
+                refresh_token=item.get("refresh_token", {"S": None})["S"],
+                expires_at_seconds=int(item["expires_at_seconds"]["N"]),
+            )
+
+        return by_resource_server
 
 
 def create_table():
@@ -120,32 +159,17 @@ def create_table():
     print("ok")
 
 
-def do_login_flow(storage: StorageAdapter):
-    auth_client.oauth2_start_flow(
-        requested_scopes=globus_sdk.GroupsClient.scopes.view_my_groups_and_memberships,
-        refresh_tokens=True,
-    )
-    authorize_url = auth_client.oauth2_get_authorize_url()
-    print(f"Please go to this URL and login:\n\n{authorize_url}\n")
-    auth_code = input("Please enter the code here: ").strip()
-    tokens = auth_client.oauth2_exchange_code_for_tokens(auth_code)
-    storage.store(tokens)
-    return tokens.by_resource_server[globus_sdk.GroupsClient.resource_server]
-
-
-def group_list(storage: StorageAdapter):
-    tokens = storage.get_token_data(globus_sdk.GroupsClient.resource_server)
-    if tokens is None:
-        tokens = do_login_flow(storage)
-
-    authorizer = globus_sdk.RefreshTokenAuthorizer(
-        tokens["refresh_token"],
-        auth_client,
-        access_token=tokens["access_token"],
-        expires_at=tokens["expires_at_seconds"],
-        on_refresh=storage.store,
-    )
-    groups_client = globus_sdk.GroupsClient(authorizer=authorizer)
+def group_list(storage: TokenStorage) -> None:
+    with globus_sdk.UserApp(
+        "dynamo-storage-example",
+        client_id=CLIENT_ID,
+        config=globus_sdk.GlobusAppConfig(token_storage=storage),
+    ) as app:
+        with globus_sdk.GroupsClient(app=app) as groups_client:
+            _print_groups(groups_client)
+
+
+def _print_groups(groups_client: globus_sdk.GroupsClient) -> None:
     print("ID,Name,Type,Session Enforcement,Roles")
     for group in groups_client.get_my_groups():
         # parse the group to get data for output
@@ -173,5 +197,5 @@ def group_list(storage: StorageAdapter):
     if args.create:
         create_table()
     else:
-        storage = DynamoDBStorageAdapter(boto_client, tablename)
+        storage = DynamoDBTokenStorage(boto_client, tablename)
         group_list(storage)

docs/examples/token_storage/index.rst

@@ -1,20 +1,31 @@
 .. _example_token_storage:
 
-Token Storage Adapters
-======================
+Token Storage objects
+=====================
 
 DynamoDB Token Storage
 ----------------------
 
-The following example demonstrates a token storage adapter which uses AWS
-DynamoDB as the backend storage mechanism.
+The following example demonstrates a token storage which uses AWS DynamoDB as
+the backend storage mechanism.
 
-Unlike the builtin adapters for JSON and sqlite, there is no capability here
-for an enumeration of all of the tokens in storage. This is because DynamoDB
-functions as a key-value store, and can efficiently map keys, but features slow
-sequential scans for enumeration. The example therefore demonstrates that
-key-value stores with limited or no capabilities for table scans can be used to
-implement the token storage interface.
+Unlike the builtin storage interfaces for JSON and sqlite, enumerating tokens in
+a DyanmoDB table-backed storage is not a desirable operation.
+DynamoDB functions as a key-value store, and can efficiently map keys,
+but features slow sequential scans for enumeration.
+
+The example implements sequential scans but also prints a noisy warning when
+that activity is triggered. An alternative implementation could raise an error
+and refuse to execute the scan.
+
+.. caution::
+
+    Raising errors on calls to get the full suite of tokens will work for many
+    use cases, but it is required by the interface so that SDK features can rely
+    on it being present.
+
+    Some capabilities, like ``GlobusApp.logout(sweep=True)`` call this method
+    and will fail if it is not implemented.
 
 .. literalinclude:: dynamodb_token_storage.py
     :caption: ``dynamodb_token_storage.py`` [:download:`download <dynamodb_token_storage.py>`]

pyproject.toml

@@ -181,12 +181,6 @@ sqlite_cache = true
 warn_unreachable = true
 warn_no_return = true
 
-[[tool.mypy.overrides]]
-module = "docs.*"
-allow_untyped_calls = true
-allow_untyped_defs = true
-ignore_missing_imports = true
-
 [tool.pylint]
 load-plugins = ["pylint.extensions.docparams"]
 accept-no-param-doc = "false"

tox.ini

@@ -65,7 +65,7 @@ commands = mypy --show-error-codes --warn-unused-ignores tests/non-pytest/mypy-i
 base = mypy
 # docs isn't really a package, so avoid errors related to it being improperly
 # treated like one (this doesn't work via config)
-commands = mypy --explicit-package-bases {posargs:docs/}
+commands = mypy --config-file docs/mypy.ini {posargs:docs/}
 
 [testenv:test-lazy-imports]
 deps = -r requirements/py{py_dot_ver}/test.txt