academy.exchange.cloud.token_store

academy/exchange/cloud/token_store.py

SafeSQLiteTokenStorage

SafeSQLiteTokenStorage(
    filepath: Path | str, *, namespace: str = "DEFAULT"
)

Bases: TokenStorage

A thread safe Globus SQLite token store.

Parameters:

• filepath (Path | str) –

  The path to a file where token data should be stored.

• namespace (str, default: 'DEFAULT' ) –

  A unique string for partitioning token data (Default: "DEFAULT").

Source code in academy/exchange/cloud/token_store.py

def __init__(
    self,
    filepath: pathlib.Path | str,
    *,
    namespace: str = 'DEFAULT',
) -> None:
    super().__init__(namespace=namespace)
    self._local_data = threading.local()
    self.filepath = filepath
    self.namespace = namespace

token_store property

token_store: SQLiteTokenStorage

Internal thread local token storage.

store_token_data_by_resource_server

store_token_data_by_resource_server(
    token_data_by_resource_server: Mapping[
        str, TokenStorageData
    ],
) -> None

Store token data for resource server(s) in the current namespace.

Parameters:

• token_data_by_resource_server (Mapping[str, TokenStorageData]) –

  mapping of resource server to token data.

Source code in academy/exchange/cloud/token_store.py

def store_token_data_by_resource_server(
    self,
    token_data_by_resource_server: Mapping[str, TokenStorageData],
) -> None:
    """Store token data for resource server(s) in the current namespace.

    Args:
        token_data_by_resource_server: mapping of resource server to
            token data.
    """
    return self.token_store.store_token_data_by_resource_server(
        token_data_by_resource_server,
    )

get_token_data_by_resource_server

get_token_data_by_resource_server() -> (
    dict[str, TokenStorageData]
)

Retrieve all token data stored in the current namespace.

Returns:

• dict[str, TokenStorageData] –

  a dict of TokenStorageData objects indexed by their resource server.

Source code in academy/exchange/cloud/token_store.py

def get_token_data_by_resource_server(self) -> dict[str, TokenStorageData]:
    """Retrieve all token data stored in the current namespace.

    Returns:
        a dict of ``TokenStorageData`` objects indexed by their
            resource server.
    """
    return self.token_store.get_token_data_by_resource_server()

remove_token_data

remove_token_data(resource_server: str) -> bool

Remove token data for a resource server in the current namespace.

Parameters:

• resource_server (str) –

  The resource server to remove token data for.

Returns:

• bool –

  True if token data was deleted, False if none was found to delete.

Source code in academy/exchange/cloud/token_store.py

def remove_token_data(self, resource_server: str) -> bool:
    """Remove token data for a resource server in the current namespace.

    Args:
        resource_server: The resource server to remove token data for.

    Returns:
        True if token data was deleted, False if none was found to delete.
    """
    return self.token_store.remove_token_data(resource_server)