Source code for structum_lab.plugins.dynaconf.core.manager

# src/structum_lab.plugins.dynaconf/manager.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods

"""Gestore di Configurazione Centrale per Structum.

Fornisce un punto di accesso centralizzato e singleton per tutte le configurazioni.
Supporta la registrazione dinamica dei builder per estendibilità.
"""

from typing import Dict, Type, Any, Union
from functools import lru_cache
import logging

from structum_lab.plugins.dynaconf.core.director import ConfigDirector
from structum_lab.plugins.dynaconf.core.exceptions import ConfigurationError
from structum_lab.plugins.dynaconf.core.builders import AbstractConfigBuilder

log = logging.getLogger(__name__)



[docs]
class ConfigManager:
    """Gestisce il caricamento, il caching e l'accesso alle configurazioni."""


[docs]
    def __init__(self) -> None:
        # Registro dinamico: nome -> Classe Builder o Istanza Builder
        self._builders: Dict[
            str, Union[Type[AbstractConfigBuilder], AbstractConfigBuilder]
        ] = {}



[docs]
    def register_builder(
        self,
        name: str,
        builder: Union[Type[AbstractConfigBuilder], AbstractConfigBuilder],
    ) -> None:
        """Registra un nuovo builder di configurazione."""
        if name in self._builders:
            log.warning(f"Overwriting builder for configuration '{name}'")
        self._builders[name] = builder

        builder_name = (
            builder.__name__
            if isinstance(builder, type)
            else builder.__class__.__name__
        )
        log.debug(f"Registered builder: {name} -> {builder_name}")



[docs]
    def load_all(self) -> None:
        """Carica forzatamente tutte le configurazioni registrate."""
        log.info("Forcing load of all registered configurations...")
        # Puliamo la cache prima di ricaricare
        self.refresh()
        for config_name in self._builders:
            try:
                self.get_config(config_name)
            except Exception as e:
                log.error(f"Failed to load configuration '{config_name}': {e}")
        log.info("All configurations loaded.")



[docs]
    @lru_cache(maxsize=None)
    def get_config(self, config_name: str) -> Any:
        """Recupera una configurazione specifica con caching lazy."""
        if config_name not in self._builders:
            log.error(f"Unsupported configuration name requested: '{config_name}'")
            raise ConfigurationError(
                f"Configuration '{config_name}' is not registered."
            )

        # Ottieni builder (classe o istanza)
        builder_or_cls = self._builders[config_name]

        # Se è una classe, istanziala. Se è un'istanza, usala direttamente.
        if isinstance(builder_or_cls, type):
            builder_instance = builder_or_cls()
        else:
            builder_instance = builder_or_cls

        # Orchestra la costruzione usando il Director
        log.debug(f"Building configuration '{config_name}'...")
        config = ConfigDirector.construct(builder_instance)

        return config



[docs]
    def refresh(self) -> None:
        """Invalida la cache interna."""
        self.get_config.cache_clear()
        log.info("Configuration cache cleared.")