|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Configuration adapter implementation for Port and Adapter pattern. |
| 3 | +
|
| 4 | +This module implements the ConfigPort interface by adapting the config module. |
| 5 | +This is the adapter that mediates all configuration access. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +from typing import Any |
| 11 | + |
| 12 | +from ...config import ( |
| 13 | + apply_config_file, |
| 14 | + get_config_schema, |
| 15 | + load_config_file, |
| 16 | +) |
| 17 | +from ...config.core.manager import config as global_config |
| 18 | +from ..ports.config_port import ConfigPort |
| 19 | + |
| 20 | + |
| 21 | +class ConfigAdapter(ConfigPort): |
| 22 | + """Adapter that implements ConfigPort by delegating to the config module. |
| 23 | + |
| 24 | + This adapter acts as a mediator between other modules and the config module, |
| 25 | + implementing the Port and Adapter (Hexagonal Architecture) pattern. |
| 26 | + """ |
| 27 | + |
| 28 | + def __init__(self, config_instance: Any = None): |
| 29 | + """Initialize the config adapter. |
| 30 | + |
| 31 | + Args: |
| 32 | + config_instance: Optional configuration instance to use. |
| 33 | + If None, uses the global config instance. |
| 34 | + """ |
| 35 | + self._config = config_instance or global_config |
| 36 | + |
| 37 | + def get(self, key: str, default: Any = None) -> Any: |
| 38 | + """Get a configuration value by key. |
| 39 | + |
| 40 | + Args: |
| 41 | + key: Configuration key |
| 42 | + default: Default value if key not found |
| 43 | + |
| 44 | + Returns: |
| 45 | + Configuration value or default |
| 46 | + """ |
| 47 | + return self._config.get(key, default) |
| 48 | + |
| 49 | + def set(self, key: str, value: Any) -> None: |
| 50 | + """Set a configuration value. |
| 51 | + |
| 52 | + Args: |
| 53 | + key: Configuration key |
| 54 | + value: Configuration value |
| 55 | + """ |
| 56 | + self._config.set(key, value) |
| 57 | + |
| 58 | + def update(self, config_dict: dict[str, Any]) -> None: |
| 59 | + """Update configuration with values from a dictionary. |
| 60 | + |
| 61 | + Args: |
| 62 | + config_dict: Dictionary of configuration values to update |
| 63 | + """ |
| 64 | + self._config.update(config_dict) |
| 65 | + |
| 66 | + def load_file(self, file_path: str) -> dict[str, Any]: |
| 67 | + """Load configuration from a file. |
| 68 | + |
| 69 | + Args: |
| 70 | + file_path: Path to configuration file |
| 71 | + |
| 72 | + Returns: |
| 73 | + Dictionary containing loaded configuration |
| 74 | + |
| 75 | + Raises: |
| 76 | + ConfigFileError: If file cannot be loaded |
| 77 | + """ |
| 78 | + return load_config_file(file_path) |
| 79 | + |
| 80 | + def apply_file( |
| 81 | + self, |
| 82 | + config_path: str | None = None, |
| 83 | + search_paths: list[str] | None = None, |
| 84 | + file_names: list[str] | None = None, |
| 85 | + ) -> bool: |
| 86 | + """Load and apply configuration from a file. |
| 87 | + |
| 88 | + Args: |
| 89 | + config_path: Explicit path to config file |
| 90 | + search_paths: List of directories to search |
| 91 | + file_names: List of file names to search for |
| 92 | + |
| 93 | + Returns: |
| 94 | + True if configuration was loaded and applied, False otherwise |
| 95 | + """ |
| 96 | + return apply_config_file( |
| 97 | + config_path=config_path, |
| 98 | + search_paths=search_paths, |
| 99 | + file_names=file_names, |
| 100 | + ) |
| 101 | + |
| 102 | + def get_schema(self) -> dict[str, Any]: |
| 103 | + """Get the JSON schema for configuration validation. |
| 104 | + |
| 105 | + Returns: |
| 106 | + JSON schema dictionary |
| 107 | + """ |
| 108 | + return get_config_schema() |
| 109 | + |
| 110 | + |
| 111 | +# Global instance for convenience (acts as the mediator) |
| 112 | +config_mediator: ConfigPort = ConfigAdapter() |
| 113 | + |
0 commit comments