Source code for structum.config.interface
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods
"""
Structum Configuration Interface.
This module defines the **Formal Contract** (Protocol) that all configuration
providers must respect. The Structum core depends *exclusively* on this
interface (Dependency Inversion), not on concrete implementations.
**Architectural Role**:
- **Decoupling**: Application code knows `ConfigProviderInterface`, not `Dynaconf`.
- **Extensibility**: Plugins (Dynaconf, Vault, Etcd) register themselves at bootstrap.
"""
import importlib
from typing import Any, Protocol
[docs]
class ConfigProviderInterface(Protocol):
"""
Protocol defining the interface for configuration providers.
This interface uses ``typing.Protocol`` to enable duck typing:
explicit inheritance is not required. Any object implementing
these methods with compatible signatures is considered a valid provider.
This approach maximizes flexibility and reduces coupling between
core and plugins.
Implementations:
- :class:`~structum.plugins.dynaconf.core.provider.DynaconfConfigProvider`
- :class:`~structum.config.manager.JSONConfigProvider` (fallback)
Example:
Using a configuration provider::
from structum.config import get_config_provider
config = get_config_provider()
# Get value with fallback
db_host = config.get("database.host", default="localhost")
# Set value
config.set[Any]("database.port", 5432)
# Check existence
if config.has("database.password"):
password = config.get("database.password")
# Persist changes
config.save()
Note:
All providers should be thread-safe and support hierarchical
key access using dot-notation (e.g., "database.pool.size").
See Also:
:func:`get_config_provider`: Retrieve the global configuration provider
:func:`set_config_provider`: Register a custom provider
"""
[docs]
def get(self, key: str, default: Any = None) -> Any:
"""
Retrieve a configuration value by key.
Supports dot-notation for nested configuration values
(e.g., "database.pool.size" accesses nested dictionaries).
Args:
key (str): Configuration key to retrieve. Supports
dot-notation for hierarchical access.
default (Any, optional): Fallback value if key doesn't exist.
Defaults to None.
Returns:
Any: The configuration value, or default if key not found.
Raises:
KeyError: If key not found and default not provided
(implementation-specific behavior).
Example:
Retrieving nested configuration::
# Config: {"database": {"host": "localhost", "port": 5432}}
host = config.get("database.host") # "localhost"
port = config.get("database.port") # 5432
# With fallback
timeout = config.get("database.timeout", 30) # 30
See Also:
:meth:`set[Any]`: Set a configuration value
:meth:`has`: Check if a key exists
"""
...
[docs]
def set(self, key: str, value: Any) -> None:
"""
Set a configuration value.
Persistence behavior depends on the concrete provider implementation.
Changes may be in-memory only until :meth:`save` is called.
Args:
key (str): Configuration key to set[Any]. Supports dot-notation
to create nested structures.
value (Any): Value to associate with the key. Can be any
JSON-serializable type.
Example:
Setting configuration values::
# Simple value
config.set[Any]("app.debug", True)
# Nested structure (creates intermediate dicts)
config.set[Any]("database.pool.size", 10)
# Complex value
config.set[Any]("servers", ["srv1", "srv2", "srv3"])
Warning:
Changes are not persisted to storage until :meth:`save` is called
(for file-based providers). In-memory providers lose changes on restart.
See Also:
:meth:`get`: Retrieve a configuration value
:meth:`save`: Persist changes to storage
"""
...
[docs]
def has(self, key: str) -> bool:
"""
Check if a configuration key exists.
Args:
key (str): Configuration key to check. Supports dot-notation.
Returns:
bool: True if the key exists, False otherwise.
Example:
Checking key existence::
if config.has("database.password"):
password = config.get("database.password")
else:
raise ValueError("Database password not configured")
Note:
A key can exist with a None value. Use :meth:`get` to
distinguish between missing keys and None values.
"""
...
[docs]
def save(self) -> None:
"""
Persist configuration changes to underlying storage.
For file-based providers, writes changes to disk.
For remote providers, may trigger a commit or synchronization.
For in-memory providers, this may be a no-op.
Raises:
IOError: If unable to write to storage (file permissions, disk full).
RuntimeError: If provider doesn't support persistence.
Example:
Saving configuration::
config.set[Any]("app.version", "2.0.0")
config.set[Any]("app.build", 123)
config.save() # Persist both changes
Warning:
Unsaved changes will be lost on process termination.
Call save() periodically for critical configuration updates.
See Also:
:meth:`reload`: Discard changes and reload from storage
"""
...
[docs]
def reload(self) -> None:
"""
Reload configuration from persistent storage.
Discards all unsaved in-memory changes and reloads the
configuration from the underlying storage source.
Raises:
IOError: If unable to read from storage.
Example:
Reloading configuration::
config.set[Any]("temp.value", 123) # In-memory change
config.reload() # Discards temp.value
# Now config reflects disk state
assert not config.has("temp.value")
Warning:
All unsaved changes will be permanently lost.
Consider calling :meth:`save` before reload if needed.
See Also:
:meth:`save`: Persist changes before reloading
"""
...
_config_provider: ConfigProviderInterface | None = None
[docs]
def get_config_provider() -> ConfigProviderInterface:
"""
Returns the global configuration provider (Lazy Loading & Dynamic).
**Architectural Role**: Singleton Accessor.
If no provider handles been registered, it automatically instantiates
the **Fallback JSON Provider** (Principle of Operational Continuity).
This ensures the application works out-of-the-box without setup.
Returns:
ConfigInterface: The active configuration provider.
"""
global _config_provider
if _config_provider is None:
# Usiamo importlib per evitare che il linter cerchi di risolvere
# staticamente dipendenze circolari o file non ancora indicizzati.
try:
module = importlib.import_module(".manager", package=__package__)
_config_provider = module.JSONConfigProvider()
except (ImportError, AttributeError) as e:
# Emergency fallback if manager.py is missing (should not happen in dist)
raise RuntimeError("Unable to load default configuration provider.") from e
pass
# Controllo finale di sicurezza
if _config_provider is None:
raise RuntimeError("Impossibile inizializzare il ConfigProvider (Critical Failure)")
return _config_provider
[docs]
def set_config_provider(
provider: ConfigProviderInterface | type[ConfigProviderInterface],
) -> None:
"""
Registers a custom configuration provider.
**Architectural Role**: Dependency Injection Root.
This function is typically called by a plugin (e.g., `structum-dynaconf`)
during the bootstrap phase. The registered provider globally replaces
the default one, seamlessly upgrading the application capabilities.
Args:
provider: Instance or Class of a provider conforming to ConfigInterface.
"""
global _config_provider
if isinstance(provider, type):
provider = provider()
_config_provider = provider