|
1 | | -import typing |
2 | 1 | from collections.abc import Callable |
| 2 | +from typing import NotRequired, TypedDict, TypeIs |
3 | 3 |
|
4 | | -_static_config: dict[str, str | None] = { |
5 | | - "data_api_root_url": None, |
6 | | - "api_key": None, |
7 | | -} |
| 4 | +import tenacity |
8 | 5 |
|
9 | | -_config_provider: Callable[[], dict[str, str | None]] | None = None |
10 | 6 |
|
| 7 | +class BubbleConfigError(RuntimeError): |
| 8 | + """Raised when Bubble API is not configured.""" |
11 | 9 |
|
12 | | -def configure(data_api_root_url: str, api_key: str) -> None: |
| 10 | + def __init__(self) -> None: |
| 11 | + super().__init__("Bubble API not configured. Call configure() first.") |
| 12 | + |
| 13 | + |
| 14 | +class _NotSet: |
| 15 | + """Sentinel for configuration values that were not provided.""" |
| 16 | + |
| 17 | + __slots__ = () |
| 18 | + |
| 19 | + def __repr__(self) -> str: |
| 20 | + return "NOT_SET" |
| 21 | + |
| 22 | + |
| 23 | +NOT_SET = _NotSet() |
| 24 | +type NotSetType = _NotSet |
| 25 | + |
| 26 | + |
| 27 | +def is_set[T](value: T | NotSetType) -> TypeIs[T]: |
| 28 | + """Type guard for checking if value is not the NOT_SET sentinel.""" |
| 29 | + return value is not NOT_SET |
| 30 | + |
| 31 | + |
| 32 | +class BubbleConfig(TypedDict): |
| 33 | + """Configuration for Bubble Data API client.""" |
| 34 | + |
| 35 | + data_api_root_url: str |
| 36 | + api_key: str |
| 37 | + retry: NotRequired[tenacity.AsyncRetrying | None] |
| 38 | + |
| 39 | + |
| 40 | +type ConfigProvider = Callable[[], BubbleConfig] |
| 41 | + |
| 42 | +_static_config: BubbleConfig | None = None |
| 43 | +_config_provider: ConfigProvider | None = None |
| 44 | + |
| 45 | + |
| 46 | +def configure( |
| 47 | + data_api_root_url: str, |
| 48 | + api_key: str, |
| 49 | + retry: tenacity.AsyncRetrying | None | NotSetType = NOT_SET, |
| 50 | +) -> None: |
13 | 51 | """Configure the Bubble Data API client with static values.""" |
14 | | - global _config_provider |
| 52 | + global _config_provider, _static_config |
15 | 53 | _config_provider = None |
16 | | - _static_config["data_api_root_url"] = data_api_root_url |
17 | | - _static_config["api_key"] = api_key |
| 54 | + _static_config = { |
| 55 | + "data_api_root_url": data_api_root_url, |
| 56 | + "api_key": api_key, |
| 57 | + } |
| 58 | + if is_set(retry): |
| 59 | + _static_config["retry"] = retry |
18 | 60 |
|
19 | 61 |
|
20 | | -def set_config_provider(provider: Callable[[], dict[str, str | None]]) -> None: |
| 62 | +def set_config_provider(provider: ConfigProvider) -> None: |
21 | 63 | """Set a provider function for dynamic configuration.""" |
22 | 64 | global _config_provider |
23 | 65 | _config_provider = provider |
24 | 66 |
|
25 | 67 |
|
26 | | -def get_config() -> typing.Mapping[str, str | None]: |
| 68 | +def get_config() -> BubbleConfig: |
27 | 69 | """Get current configuration from provider if set, otherwise static config.""" |
28 | 70 | if _config_provider is not None: |
29 | 71 | return _config_provider() |
| 72 | + if _static_config is None: |
| 73 | + raise BubbleConfigError |
30 | 74 | return _static_config |
0 commit comments