Skip to content

proxystore.globus.app

Create GlobusApp instances.

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 ProxyStore home directory (see home_dir()).

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

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

Returns:

Source code in proxystore/globus/app.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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 ProxyStore home directory
            (see [`home_dir()`][proxystore.utils.environment.home_dir]).
        namespace: Optional namespace to use within the database for
            partitioning token data.

    Returns:
        Token storage.
    """
    if filepath is None:
        filepath = os.path.join(home_dir(), _TOKENS_FILE)
    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 PROXYSTORE_GLOBUS_CLIENT_ID and the secret to PROXYSTORE_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 proxystore/globus/app.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 `PROXYSTORE_GLOBUS_CLIENT_ID` and
    the secret to `PROXYSTORE_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[PROXYSTORE_GLOBUS_CLIENT_ID_ENV_NAME]
        client_secret = os.environ[PROXYSTORE_GLOBUS_CLIENT_SECRET_ENV_NAME]
    except KeyError as e:
        raise ValueError(
            f'Both {PROXYSTORE_GLOBUS_CLIENT_ID_ENV_NAME} and '
            f'{PROXYSTORE_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 proxystore/globus/app.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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()`][proxystore.globus.app.get_client_app]. Otherwise
    returns a [`UserApp`][globus_sdk.UserApp] using
    [`get_user_app()`][proxystore.globus.app.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 proxystore/globus/app.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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()`][proxystore.globus.app.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 proxystore/globus/app.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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=PROXYSTORE_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 PROXYSTORE_GLOBUS_CLIENT_ID and PROXYSTORE_GLOBUS_CLIENT_SECRET are set.

Source code in proxystore/globus/app.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 `PROXYSTORE_GLOBUS_CLIENT_ID` and \
        `PROXYSTORE_GLOBUS_CLIENT_SECRET` are set.
    """
    try:
        get_client_credentials_from_env()
    except ValueError:
        return False
    else:
        return True