-
Notifications
You must be signed in to change notification settings - Fork 11
Expand file tree
/
Copy pathconfiguration.py
More file actions
368 lines (307 loc) · 13.5 KB
/
configuration.py
File metadata and controls
368 lines (307 loc) · 13.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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
88
89
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
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
import contextlib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Optional
from galileo.__future__.shared.exceptions import ConfigurationError
from galileo.config import GalileoPythonConfig
from galileo.utils.logging import get_logger
logger = get_logger(__name__)
@dataclass(frozen=True)
class ConfigKey:
"""
Metadata for a configuration key.
Defines a single configuration option with its properties, validation rules,
and relationship to environment variables. Used by the ConfigurationMeta
metaclass to enable dynamic attribute access.
Attributes
----------
name : str
The attribute name used in the Configuration class (e.g., "galileo_api_key").
env_var : str
The corresponding environment variable name (e.g., "GALILEO_API_KEY").
description : str
Human-readable description of the configuration key's purpose.
required : bool
Whether this key must be set for the configuration to be considered complete.
Default: False.
sensitive : bool
Whether this key contains sensitive information (e.g., API keys, passwords).
Sensitive values are masked in get_configuration() output. Default: False.
default : Any
The default value if no explicit value or environment variable is set.
Default: None.
value_type : type
The expected Python type for this configuration value. Default: str.
parser : Optional[Callable[[str], Any]]
Optional function to convert string values from environment variables
to the appropriate type. Default: None (no conversion).
"""
name: str
env_var: str
description: str
required: bool = False
sensitive: bool = False
default: Any = None
value_type: type = str
parser: Optional[Callable[[str], Any]] = None
_CONFIGURATION_KEYS = [
ConfigKey(
name="galileo_api_key",
env_var="GALILEO_API_KEY",
description="API key for authenticating with Galileo",
required=True,
sensitive=True,
),
ConfigKey(
name="console_url", env_var="GALILEO_CONSOLE_URL", description="URL of the Galileo console", required=True
),
ConfigKey(
name="openai_api_key",
env_var="OPENAI_API_KEY",
description="OpenAI API key for interoperability with OpenAI SDK",
sensitive=True,
),
ConfigKey(name="default_project_name", env_var="GALILEO_PROJECT", description="Default project name"),
ConfigKey(name="default_project_id", env_var="GALILEO_PROJECT_ID", description="Default project ID"),
ConfigKey(name="default_logstream_name", env_var="GALILEO_LOG_STREAM", description="Default log stream name"),
ConfigKey(name="default_logstream_id", env_var="GALILEO_LOG_STREAM_ID", description="Default log stream ID"),
ConfigKey(
name="logging_disabled",
env_var="GALILEO_LOGGING_DISABLED",
description="Disable all logging to Galileo",
default=False,
value_type=bool,
parser=lambda v: v.lower() in ("true", "1", "t", "yes"),
),
]
_KEYS_BY_NAME = {key.name: key for key in _CONFIGURATION_KEYS}
class ConfigurationMeta(type):
"""
Metaclass for dynamic attribute handling based on CONFIGURATION_KEYS.
This metaclass enables the Configuration class to provide dynamic attribute access
with automatic resolution from multiple sources. When accessing a configuration
attribute (e.g., `Configuration.galileo_api_key`), the metaclass:
1. Checks if the attribute is in _CONFIGURATION_KEYS
2. Resolves the value from: explicit value → environment variable → .env file → default
3. Applies type conversion and validation if a parser is defined
4. Returns the resolved value
When setting a configuration attribute (e.g., `Configuration.galileo_api_key = "key"`),
the metaclass:
1. Stores the value internally (with underscore prefix: `_galileo_api_key`)
2. Automatically updates the corresponding environment variable
3. Ensures compatibility with libraries that read from os.environ
This pattern provides:
- Extensibility: New configuration keys can be added to _CONFIGURATION_KEYS
- Consistency: All configuration uses the same resolution pattern
- Compatibility: Environment variables are kept in sync for third-party libraries
- Transparency: Configuration sources are clearly prioritized
Reference: https://docs.python.org/3/reference/datamodel.html#customizing-attribute-access
"""
def __getattribute__(cls, name: str) -> Any:
"""
Get configuration attribute value with automatic source resolution.
Resolution order: explicit value → env var → .env file → default.
Args:
name: The attribute name to retrieve.
Returns
-------
The resolved configuration value, or the attribute itself if not a config key.
"""
if name.startswith("_") or name in (
"connect",
"reset",
"is_configured",
"get_configuration",
"get_key_info",
"get_keys_by_category",
):
return super().__getattribute__(name)
if name in _KEYS_BY_NAME:
key = _KEYS_BY_NAME[name]
super().__getattribute__("_load_env_file")()
internal_name = f"_{name}"
try:
explicit_value = super().__getattribute__(internal_name)
if explicit_value is not None:
return explicit_value
except AttributeError:
pass
env_value = os.environ.get(key.env_var)
if env_value is not None:
if key.parser:
return key.parser(env_value)
return env_value
return key.default
return super().__getattribute__(name)
def __setattr__(cls, name: str, value: Any) -> None:
"""
Set configuration attribute and sync to environment variable.
When setting a configuration key, the value is:
1. Stored internally with underscore prefix (e.g., `_galileo_api_key`)
2. Synced to the corresponding environment variable (e.g., `GALILEO_API_KEY`)
This ensures that both the Configuration class and environment variables
remain consistent, maintaining compatibility with third-party libraries.
Args:
name: The attribute name to set.
value: The value to assign to the attribute.
"""
if name in _KEYS_BY_NAME:
key = _KEYS_BY_NAME[name]
internal_name = f"_{name}"
super().__setattr__(internal_name, value)
if value is not None:
if key.value_type is bool:
os.environ[key.env_var] = str(value).lower()
else:
os.environ[key.env_var] = str(value)
else:
super().__setattr__(name, value)
class Configuration(metaclass=ConfigurationMeta):
"""
Single source of truth for SDK configuration.
This class uses a metaclass pattern to provide dynamic attribute access to configuration
keys defined in _CONFIGURATION_KEYS. Each configuration key can be accessed as a class
attribute, with values resolved in the following priority order:
1. Explicitly set value (via `Configuration.key = value`)
2. Environment variable (e.g., GALILEO_API_KEY)
3. .env file (loaded automatically on first access)
4. Default value (defined in the key configuration)
The metaclass automatically:
- Syncs attribute assignments to environment variables
- Loads .env files on first configuration access
- Provides type conversion and validation via parsers
Attributes
----------
Configuration attributes are dynamically provided based on _CONFIGURATION_KEYS:
galileo_api_key (str): API key for Galileo authentication (sensitive)
console_url (str): URL of the Galileo console
openai_api_key (str): OpenAI API key for SDK interoperability (sensitive)
default_project_name (str): Default project name
default_project_id (str): Default project ID
default_logstream_name (str): Default log stream name
default_logstream_id (str): Default log stream ID
logging_disabled (bool): Disable all logging to Galileo
Examples
--------
Reading configuration values:
```python
# Access via class attribute (reads from env vars, .env, or defaults)
api_key = Configuration.galileo_api_key
url = Configuration.console_url
```
Setting configuration values:
```python
# Set explicitly (also updates environment variables)
Configuration.galileo_api_key = "your-api-key"
Configuration.console_url = "https://console.galileo.ai"
```
Checking and connecting:
```python
# Check if required configuration is present
if Configuration.is_configured():
Configuration.connect()
```
Getting all configuration:
```python
# Get all configuration values (sensitive values are masked)
config = Configuration.get_configuration()
print(config["galileo_api_key"]) # Output: "***"
print(config["console_url"]) # Output: actual URL
```
Resetting configuration:
```python
# Clear all configuration values and environment variables
Configuration.reset()
```
Notes
-----
- The Configuration class should not be instantiated; use it as a static class
- Direct attribute access (e.g., `Configuration.galileo_api_key`) is the recommended pattern
- The `get_configuration()` method masks sensitive values for safe display/logging
- Setting an attribute automatically updates the corresponding environment variable
- This design maintains compatibility with third-party libraries expecting env vars
"""
_env_loaded: bool = False
@classmethod
def _load_env_file(cls) -> None:
"""Load .env file if present. Called automatically on first access."""
if cls._env_loaded:
return
env_file = Path(".env")
if env_file.exists():
try:
with open(env_file) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key not in os.environ:
os.environ[key] = value
except Exception as e:
logger.debug(f"Failed to parse .env file: {e}")
cls._env_loaded = True
@classmethod
def connect(cls) -> None:
"""Validate configuration and connect to Galileo."""
cls._load_env_file()
if not cls.console_url:
raise ConfigurationError(
"Galileo console URL is required. Set Configuration.console_url or GALILEO_CONSOLE_URL."
)
if not cls.galileo_api_key:
raise ConfigurationError(
"Galileo API key is required. Set Configuration.galileo_api_key or GALILEO_API_KEY."
)
logger.info("Validating Galileo configuration and connectivity...")
try:
GalileoPythonConfig.get()
logger.info("Successfully connected to Galileo")
except Exception as e:
error_msg = str(e)
logger.error(f"Failed to connect to Galileo: {error_msg}")
error_lower = error_msg.lower()
if ("api" in error_lower and "key" in error_lower) or "auth" in error_lower:
raise ConfigurationError(f"Authentication failed: {error_msg}") from e
if "url" in error_lower or "connection" in error_lower:
raise ConfigurationError(f"Connection failed: {error_msg}") from e
raise ConfigurationError(f"Configuration validation failed: {error_msg}") from e
@classmethod
def reset(cls) -> None:
"""Reset all configuration values and clear environment variables."""
for key in _CONFIGURATION_KEYS:
with contextlib.suppress(AttributeError):
delattr(cls, f"_{key.name}")
if key.env_var in os.environ:
del os.environ[key.env_var]
cls._env_loaded = False
try:
if GalileoPythonConfig._instance is not None:
GalileoPythonConfig._instance.reset()
except Exception as e:
logger.debug(f"Could not reset GalileoPythonConfig instance: {e}")
@classmethod
def is_configured(cls) -> bool:
"""Check if all required configuration keys are set."""
cls._load_env_file()
return all(getattr(cls, key.name) for key in _CONFIGURATION_KEYS if key.required)
@classmethod
def get_configuration(cls) -> dict[str, Any]:
"""Get all configuration values (sensitive values are masked)."""
cls._load_env_file()
result: dict[str, Any] = {}
for key in _CONFIGURATION_KEYS:
value = getattr(cls, key.name)
if key.sensitive and value:
result[key.name] = "***"
elif value is None:
result[key.name] = "Not set"
else:
result[key.name] = value
result["is_configured"] = cls.is_configured()
result["env_file_loaded"] = cls._env_loaded
return result