Skip to content

academy.exchange.cloud.login

Create GlobusApp instances.

Taken from: https://github.com/proxystore/proxystore/blob/e296f937d913cae02f87ab35112fd4c8351860b0/proxystore/globus/app.py

get_token_storage 

get_token_storage(
    filepath: str | Path | None = None,
    *,
    namespace: str = "DEFAULT"
) -> SQLiteTokenStorage

Create token storage adapter.

Parameters:

  • filepath (str | Path | None, default: None ) –

    Name of the database file. If not provided, defaults to a file in the Academy home directory specified by ACADEMY_HOME.

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

    Optional namespace to use within the database for partitioning token data.

Returns:

Source code in academy/exchange/cloud/login.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_token_storage(
    filepath: str | pathlib.Path | None = None,
    *,
    namespace: str = 'DEFAULT',
) -> SQLiteTokenStorage:
    """Create token storage adapter.

    Args:
        filepath: Name of the database file. If not provided, defaults to a
            file in the Academy home directory specified by ACADEMY_HOME.
        namespace: Optional namespace to use within the database for
            partitioning token data.

    Returns:
        Token storage.
    """
    if filepath is None:
        default = os.path.join(
            os.path.expanduser('~/local/share'),
            _APP_NAME,
        )
        basepath = os.environ.get('ACADEMY_HOME', default=default)
        filepath = os.path.join(basepath, _TOKENS_FILE)

    print(filepath)
    filepath = pathlib.Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)
    return SQLiteTokenStorage(filepath, namespace=namespace)

get_client_credentials_from_env 

get_client_credentials_from_env() -> tuple[str, str]

Read the Globus Client ID and secret from the environment.

The Client ID should be set to ACADEMY_GLOBUS_CLIENT_ID and the secret to ACADEMY_GLOBUS_CLIENT_SECRET.

Note

This function performs no validation on the values of the variables.

Returns:

  • tuple[str, str] –

    Tuple containing the client ID and secret.

Raises:

  • ValueError –

    if one of the environment variables is set.

Source code in academy/exchange/cloud/login.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def get_client_credentials_from_env() -> tuple[str, str]:
    """Read the Globus Client ID and secret from the environment.

    The Client ID should be set to `ACADEMY_GLOBUS_CLIENT_ID` and
    the secret to `ACADEMY_GLOBUS_CLIENT_SECRET`.

    Note:
        This function performs no validation on the values of the variables.

    Returns:
        Tuple containing the client ID and secret.

    Raises:
        ValueError: if one of the environment variables is set.
    """
    try:
        client_id = os.environ[ACADEMY_GLOBUS_CLIENT_ID_ENV_NAME]
        client_secret = os.environ[ACADEMY_GLOBUS_CLIENT_SECRET_ENV_NAME]
    except KeyError as e:
        raise ValueError(
            f'Both {ACADEMY_GLOBUS_CLIENT_ID_ENV_NAME} and '
            f'{ACADEMY_GLOBUS_CLIENT_SECRET_ENV_NAME} must be set to '
            'use a client identity. Either set both environment variables '
            'or unset both to use the normal login flow.',
        ) from e

    return client_id, client_secret

get_globus_app 

get_globus_app() -> GlobusApp

Get a Globus App based on the environment.

If a client ID and secret are set in the environment, returns a ClientApp using get_client_app(). Otherwise returns a UserApp using get_user_app().

Returns:

  • GlobusApp –

    Initialized app.

Source code in academy/exchange/cloud/login.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def get_globus_app() -> GlobusApp:
    """Get a Globus App based on the environment.

    If a client ID and secret are set in the environment, returns a
    [`ClientApp`][globus_sdk.ClientApp] using
    [`get_client_app()`][academy.exchange.cloud.login.get_client_app].
    Otherwise returns a [`UserApp`][globus_sdk.UserApp] using
    [`get_user_app()`][academy.exchange.cloud.login.get_user_app].

    Returns:
        Initialized app.
    """
    if is_client_login():
        return get_client_app()
    return get_user_app()

get_client_app 

get_client_app(
    client_id: str | None = None,
    client_secret: str | None = None,
) -> ClientApp

Get a Client Globus App.

Parameters:

  • client_id (str | None, default: None ) –

    Client ID. If one or both of the client_id and client_secret are not provided, the values will be read from the environment using get_client_credentials_from_env().

  • client_secret (str | None, default: None ) –

    Client secret. See above.

Returns:

  • ClientApp –

    Initialized app.

Source code in academy/exchange/cloud/login.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def get_client_app(
    client_id: str | None = None,
    client_secret: str | None = None,
) -> ClientApp:
    """Get a Client Globus App.

    Args:
        client_id: Client ID. If one or both of the `client_id` and
            `client_secret` are not provided, the values will be read from
            the environment using
            [`get_client_credentials_from_env()`][academy.exchange.cloud.login.get_client_credentials_from_env].
        client_secret: Client secret. See above.

    Returns:
        Initialized app.
    """
    if client_id is None or client_secret is None:
        client_id, client_secret = get_client_credentials_from_env()

    config = GlobusAppConfig(
        token_storage=get_token_storage(),
        request_refresh_tokens=True,
    )

    return ClientApp(
        app_name=_APP_NAME,
        client_id=client_id,
        client_secret=client_secret,
        config=config,
    )

get_user_app 

get_user_app() -> UserApp

Get a User Globus App.

The UserApp will automatically perform an interactive flow with the user as needed.

Returns:

  • UserApp –

    Initialized app.

Source code in academy/exchange/cloud/login.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def get_user_app() -> UserApp:
    """Get a User Globus App.

    The [`UserApp`][globus_sdk.UserApp] will
    automatically perform an interactive flow with the user as needed.

    Returns:
        Initialized app.
    """
    config = GlobusAppConfig(
        login_flow_manager=_CustomLoginFlowManager,
        token_storage=get_token_storage(),
        request_refresh_tokens=True,
    )

    return UserApp(
        app_name=_APP_NAME,
        client_id=ACADEMY_GLOBUS_CLIENT_ID,
        config=config,
    )

is_client_login 

is_client_login() -> bool

Check if Globus client identity environment variables are set.

Based on the Globus Compute SDK's is_client_login().

Returns:

  • bool –

    True if ACADEMY_GLOBUS_CLIENT_ID and ACADEMY_GLOBUS_CLIENT_SECRET are set.

Source code in academy/exchange/cloud/login.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def is_client_login() -> bool:
    """Check if Globus client identity environment variables are set.

    Based on the Globus Compute SDK's
    [`is_client_login()`](https://github.com/funcx-faas/funcX/blob/8f5b59075ae6f8e8b8b13fe1b91430271f4e0c3c/compute_sdk/globus_compute_sdk/sdk/login_manager/client_login.py#L24-L38){target=_blank}.

    Returns:
        `True` if `ACADEMY_GLOBUS_CLIENT_ID` and \
        `ACADEMY_GLOBUS_CLIENT_SECRET` are set.
    """
    try:
        get_client_credentials_from_env()
    except ValueError:
        return False
    else:
        return True

get_auth_headers 

get_auth_headers(
    method: Literal["globus"] | None, **kwargs: Any
) -> dict[str, str]

Client utility method to perform authentication and get headers.

Source code in academy/exchange/cloud/login.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def get_auth_headers(
    method: Literal['globus'] | None,
    **kwargs: Any,
) -> dict[str, str]:
    """Client utility method to perform authentication and get headers."""
    if method is None:
        return {}
    elif method == 'globus':
        app = get_globus_app()
        app.add_scope_requirements(
            {
                AcademyExchangeScopes.resource_server: [
                    AcademyExchangeScopes.academy_exchange,
                ],
            },
        )
        try:
            authorizer = app.get_authorizer(
                AcademyExchangeScopes.resource_server,
            )
        except TokenValidationError:
            raise SystemExit(1) from None

        bearer = authorizer.get_authorization_header()
        assert bearer is not None
        return {'Authorization': bearer}
    else:
        raise AssertionError('Unreachable.')