Source code for structum_lab.config.interface
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods
"""
Interfaccia del sistema di configurazione di Structum.
Questo modulo definisce il contratto formale che tutti i provider di
configurazione devono rispettare. Il core di Structum dipende esclusivamente
da questa interfaccia e non da implementazioni concrete.
L’architettura è pensata per essere estendibile tramite plugin: provider
avanzati (Dynaconf, database, servizi remoti, ecc.) possono essere registrati
in fase di bootstrap senza modificare il core.
"""
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_lab.plugins.dynaconf.core.provider.DynaconfConfigProvider`
- :class:`~structum_lab.config.manager.JSONConfigProvider` (fallback)
Example:
Using a configuration provider::
from structum_lab.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("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`: 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. 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("app.debug", True)
# Nested structure (creates intermediate dicts)
config.set("database.pool.size", 10)
# Complex value
config.set("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("app.version", "2.0.0")
config.set("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("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:
"""
Restituisce il provider di configurazione globale (Caricamento Lazy e Dinamico).
Se nessun provider è stato registrato, viene istanziato automaticamente
il provider JSON di fallback basato sulla standard library.
Returns:
Un’implementazione di ConfigInterface.
"""
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:
# Fallback di emergenza se il file manager.py non esiste
raise RuntimeError(
"Impossibile caricare il provider di configurazione di default."
) 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:
"""
Registra un provider di configurazione personalizzato.
Questa funzione viene tipicamente chiamata da un plugin in fase
di inizializzazione. Il provider registrato sostituisce globalmente
quello precedente.
Args:
provider: Istanza o classe di un provider compatibile con ConfigInterface.
"""
global _config_provider
if isinstance(provider, type):
provider = provider()
_config_provider = provider