Collect per-provider ``bearer_token_provider`` callbacks keyed by provider name. The singular, whole-session ``provider`` uses the implicit ``_DEFAULT_BEARER_TOKEN_PROVIDER_NAME``; ``providers`` entries use their own ``name``. The callbacks are never serialized — the wire conversion emi
(
provider: ProviderConfig | None,
providers: list[NamedProviderConfig] | None,
)
| 180 | |
| 181 | |
| 182 | def _collect_bearer_token_callbacks( |
| 183 | provider: ProviderConfig | None, |
| 184 | providers: list[NamedProviderConfig] | None, |
| 185 | ) -> dict[str, BearerTokenProvider]: |
| 186 | """Collect per-provider ``bearer_token_provider`` callbacks keyed by provider name. |
| 187 | |
| 188 | The singular, whole-session ``provider`` uses the implicit |
| 189 | ``_DEFAULT_BEARER_TOKEN_PROVIDER_NAME``; ``providers`` entries use their own |
| 190 | ``name``. The callbacks are never serialized — the wire conversion emits |
| 191 | ``hasBearerTokenProvider: true`` instead and the runtime calls back over |
| 192 | ``providerToken.getToken``. |
| 193 | """ |
| 194 | callbacks: dict[str, BearerTokenProvider] = {} |
| 195 | if provider is not None: |
| 196 | singular = provider.get("bearer_token_provider") |
| 197 | if singular is not None: |
| 198 | callbacks[_DEFAULT_BEARER_TOKEN_PROVIDER_NAME] = singular |
| 199 | if providers: |
| 200 | for named in providers: |
| 201 | callback = named.get("bearer_token_provider") |
| 202 | if callback is not None: |
| 203 | callbacks[named["name"]] = callback |
| 204 | return callbacks |
| 205 | |
| 206 | |
| 207 | def _validate_session_fs_config(config: SessionFsConfig) -> None: |
no test coverage detected
searching dependent graphs…