Python YAML & Config Files: Managing Application Configuration

# Install PyYAML
pip install pyyaml

import yaml

# Sample YAML string
yaml_str = """
# Application configuration
database:
  host: localhost
  port: 5432
  name: myapp
  user: admin
  password: ${DB_PASSWORD}

redis:
  host: redis.example.com
  port: 6379
  db: 0

feature_flags:
  enable_new_ui: true
  enable_analytics: false
  max_users: 1000

logging:
  level: INFO
  format: json
  outputs:
    - console
    - file: /var/log/app.log
"""

# Parse YAML string
config = yaml.safe_load(yaml_str)
print(f"Database host: {config['database']['host']}")
print(f"Redis port: {config['redis']['port']}")
print(f"Feature flags: {config['feature_flags']}")

# Load from file
with open('config.yaml', 'r') as f:
    config = yaml.safe_load(f)

# Write YAML to file
with open('output.yaml', 'w') as f:
    yaml.dump(config, f, default_flow_style=False, indent=2)

# config/base.yaml
# Base configuration shared across all environments
app:
  name: MyApplication
  version: 1.0.0
database:
  pool_size: 10
  timeout: 30
logging:
  level: INFO
  format: json

# config/development.yaml
database:
  host: localhost
  user: dev_user
  password: dev_password
logging:
  level: DEBUG

# config/production.yaml
database:
  host: prod-db.example.com
  user: prod_user
  password: ${PROD_DB_PASSWORD}
logging:
  level: WARNING

import yaml
import os
from pathlib import Path

def load_config(env='development'):
    """Load configuration with environment-specific overrides"""
    config_dir = Path(__file__).parent / 'config'
    
    # Load base config
    with open(config_dir / 'base.yaml', 'r') as f:
        config = yaml.safe_load(f)
    
    # Load environment-specific config
    env_file = config_dir / f'{env}.yaml'
    if env_file.exists():
        with open(env_file, 'r') as f:
            env_config = yaml.safe_load(f)
            # Deep merge (simple version)
            deep_update(config, env_config)
    
    # Replace environment variables
    config = replace_env_vars(config)
    
    return config

def deep_update(base, updates):
    """Recursively update nested dictionaries"""
    for key, value in updates.items():
        if isinstance(value, dict) and key in base and isinstance(base[key], dict):
            deep_update(base[key], value)
        else:
            base[key] = value

def replace_env_vars(obj):
    """Replace ${ENV_VAR} patterns with environment variables"""
    if isinstance(obj, dict):
        return {k: replace_env_vars(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [replace_env_vars(item) for item in obj]
    elif isinstance(obj, str):
        import re
        pattern = r'\$\{([^}]+)\}'
        def replacer(match):
            env_var = match.group(1)
            return os.environ.get(env_var, '')
        return re.sub(pattern, replacer, obj)
    else:
        return obj

from typing import Any, Dict, List

class ConfigValidator:
    """Validate configuration structure and types"""
    
    def __init__(self, schema: Dict[str, Any]):
        self.schema = schema
    
    def validate(self, config: Dict[str, Any]) -> List[str]:
        """Validate config against schema, return list of errors"""
        errors = []
        self._validate_dict(config, self.schema, errors, [])
        return errors
    
    def _validate_dict(self, config: Dict, schema: Dict, errors: List, path: List):
        for key, rules in schema.items():
            current_path = path + [key]
            
            # Check required
            if rules.get('required', False) and key not in config:
                errors.append(f"Missing required field: {'.'.join(current_path)}")
                continue
            
            if key not in config:
                continue
            
            value = config[key]
            
            # Check type
            expected_type = rules.get('type')
            if expected_type and not isinstance(value, expected_type):
                errors.append(f"Field {'.'.join(current_path)} should be {expected_type.__name__}, got {type(value).__name__}")
            
            # Check nested schema
            if 'schema' in rules and isinstance(value, dict):
                self._validate_dict(value, rules['schema'], errors, current_path)
            
            # Check choices
            if 'choices' in rules and value not in rules['choices']:
                errors.append(f"Field {'.'.join(current_path)} must be one of {rules['choices']}, got {value}")
            
            # Check min/max for numbers
            if isinstance(value, (int, float)):
                if 'min' in rules and value < rules['min']:
                    errors.append(f"Field {'.'.join(current_path)} must be >= {rules['min']}, got {value}")
                if 'max' in rules and value > rules['max']:
                    errors.append(f"Field {'.'.join(current_path)} must be <= {rules['max']}, got {value}")
            
            # Check pattern for strings
            if isinstance(value, str) and 'pattern' in rules:
                import re
                if not re.match(rules['pattern'], value):
                    errors.append(f"Field {'.'.join(current_path)} does not match pattern {rules['pattern']}")


# Example schema
SCHEMA = {
    'app': {
        'required': True,
        'type': dict,
        'schema': {
            'name': {'type': str, 'required': True},
            'version': {'type': str, 'pattern': r'^\d+\.\d+\.\d+$', 'required': True}
        }
    },
    'database': {
        'required': True,
        'type': dict,
        'schema': {
            'host': {'type': str, 'required': True},
            'port': {'type': int, 'required': True, 'min': 1, 'max': 65535},
            'name': {'type': str, 'required': True},
            'user': {'type': str, 'required': True},
            'password': {'type': str, 'required': True}
        }
    },
    'logging': {
        'type': dict,
        'schema': {
            'level': {'type': str, 'choices': ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], 'default': 'INFO'},
            'format': {'type': str, 'choices': ['json', 'text'], 'default': 'json'}
        }
    }
}

#!/usr/bin/env python3
"""
config_manager.py - Complete configuration management system
"""

import os
import yaml
import json
import argparse
from pathlib import Path
from typing import Any, Dict, Optional
from dataclasses import dataclass, field
from functools import lru_cache

@dataclass
class AppConfig:
    """Application configuration dataclass"""
    name: str
    version: str
    environment: str
    debug: bool = False
    
    database_host: str = 'localhost'
    database_port: int = 5432
    database_name: str = 'app'
    database_user: str = 'postgres'
    database_password: str = ''
    
    redis_host: str = 'localhost'
    redis_port: int = 6379
    redis_db: int = 0
    
    api_key: str = ''
    api_secret: str = ''
    
    log_level: str = 'INFO'
    log_format: str = 'json'
    
    max_connections: int = 100
    request_timeout: int = 30
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any], environment: str = 'development') -> 'AppConfig':
        """Create AppConfig from dictionary"""
        app_config = data.get('app', {})
        db_config = data.get('database', {})
        redis_config = data.get('redis', {})
        auth_config = data.get('auth', {})
        logging_config = data.get('logging', {})
        performance_config = data.get('performance', {})
        
        return cls(
            name=app_config.get('name', 'MyApp'),
            version=app_config.get('version', '1.0.0'),
            environment=environment,
            debug=data.get('debug', environment == 'development'),
            
            database_host=db_config.get('host', 'localhost'),
            database_port=db_config.get('port', 5432),
            database_name=db_config.get('name', 'app'),
            database_user=db_config.get('user', 'postgres'),
            database_password=db_config.get('password', ''),
            
            redis_host=redis_config.get('host', 'localhost'),
            redis_port=redis_config.get('port', 6379),
            redis_db=redis_config.get('db', 0),
            
            api_key=auth_config.get('api_key', ''),
            api_secret=auth_config.get('api_secret', ''),
            
            log_level=logging_config.get('level', 'INFO'),
            log_format=logging_config.get('format', 'json'),
            
            max_connections=performance_config.get('max_connections', 100),
            request_timeout=performance_config.get('request_timeout', 30),
        )
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary"""
        return {
            'app': {
                'name': self.name,
                'version': self.version
            },
            'environment': self.environment,
            'debug': self.debug,
            'database': {
                'host': self.database_host,
                'port': self.database_port,
                'name': self.database_name,
                'user': self.database_user,
                'password': self.database_password
            },
            'redis': {
                'host': self.redis_host,
                'port': self.redis_port,
                'db': self.redis_db
            },
            'auth': {
                'api_key': self.api_key,
                'api_secret': self.api_secret
            },
            'logging': {
                'level': self.log_level,
                'format': self.log_format
            },
            'performance': {
                'max_connections': self.max_connections,
                'request_timeout': self.request_timeout
            }
        }
    
    def get_database_url(self) -> str:
        """Get PostgreSQL connection URL"""
        password = f':{self.database_password}' if self.database_password else ''
        return f"postgresql://{self.database_user}{password}@{self.database_host}:{self.database_port}/{self.database_name}"
    
    def get_redis_url(self) -> str:
        """Get Redis connection URL"""
        return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"
    
    def __repr__(self):
        return f"<AppConfig name={self.name} env={self.environment} debug={self.debug}>"


class ConfigLoader:
    """Load and manage configuration from multiple sources"""
    
    def __init__(self, config_dir: Optional[Path] = None):
        self.config_dir = config_dir or Path(__file__).parent / 'config'
        self._config: Optional[AppConfig] = None
    
    def load_yaml(self, filepath: Path) -> Dict[str, Any]:
        """Load YAML file"""
        with open(filepath, 'r') as f:
            return yaml.safe_load(f) or {}
    
    def load_json(self, filepath: Path) -> Dict[str, Any]:
        """Load JSON file"""
        with open(filepath, 'r') as f:
            return json.load(f)
    
    def load_from_env(self) -> Dict[str, Any]:
        """Load configuration from environment variables"""
        config = {}
        
        # APP_ prefixed variables
        for key, value in os.environ.items():
            if key.startswith('APP_'):
                config_key = key[4:].lower()
                config[config_key] = value
            
            # Nested configs: DATABASE_HOST -> database.host
            if '_' in key:
                parts = key.lower().split('_')
                if len(parts) >= 2:
                    section = parts[0]
                    field = '_'.join(parts[1:])
                    if section not in config:
                        config[section] = {}
                    config[section][field] = value
        
        return config
    
    @lru_cache(maxsize=1)
    def get_config(self, environment: Optional[str] = None) -> AppConfig:
        """Get application configuration (cached)"""
        env = environment or os.environ.get('APP_ENV', 'development')
        
        # Load base config
        base_config = {}
        base_file = self.config_dir / 'base.yaml'
        if base_file.exists():
            base_config = self.load_yaml(base_file)
        
        # Load environment-specific config
        env_file = self.config_dir / f'{env}.yaml'
        if env_file.exists():
            env_config = self.load_yaml(env_file)
            self._deep_update(base_config, env_config)
        
        # Override with environment variables
        env_overrides = self.load_from_env()
        self._deep_update(base_config, env_overrides)
        
        # Replace environment variable placeholders
        base_config = self._replace_env_vars(base_config)
        
        # Create AppConfig
        self._config = AppConfig.from_dict(base_config, env)
        
        # Validate configuration
        self._validate()
        
        return self._config
    
    def _deep_update(self, base: Dict, updates: Dict):
        """Recursively update dictionary"""
        for key, value in updates.items():
            if isinstance(value, dict) and key in base and isinstance(base[key], dict):
                self._deep_update(base[key], value)
            else:
                base[key] = value
    
    def _replace_env_vars(self, obj: Any) -> Any:
        """Replace ${VAR} placeholders"""
        if isinstance(obj, dict):
            return {k: self._replace_env_vars(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [self._replace_env_vars(item) for item in obj]
        elif isinstance(obj, str):
            import re
            def replacer(match):
                env_var = match.group(1)
                return os.environ.get(env_var, match.group(0))
            return re.sub(r'\$\{([^}]+)\}', replacer, obj)
        return obj
    
    def _validate(self):
        """Validate configuration"""
        if not self._config:
            raise ValueError("Configuration not loaded")
        
        if self._config.environment == 'production':
            # Production validation
            if not self._config.database_password:
                raise ValueError("Production requires database password")
            if not self._config.api_key:
                raise ValueError("Production requires API key")
            if self._config.debug:
                raise ValueError("Production should not have debug=True")
            if self._config.log_level == 'DEBUG':
                raise ValueError("Production should not use DEBUG logging")
    
    def reload(self):
        """Reload configuration (clear cache)"""
        self.get_config.cache_clear()
        self._config = None


def main():
    parser = argparse.ArgumentParser(description='Configuration Manager')
    parser.add_argument('--env', '-e', default='development', help='Environment name')
    parser.add_argument('--show', action='store_true', help='Show configuration')
    parser.add_argument('--export', help='Export config to file')
    parser.add_argument('--validate', action='store_true', help='Validate configuration')
    
    args = parser.parse_args()
    
    loader = ConfigLoader()
    
    try:
        config = loader.get_config(args.env)
        
        if args.show:
            print(f"\n📋 Configuration for: {config.environment}")
            print("="*50)
            print(f"App: {config.name} v{config.version}")
            print(f"Debug: {config.debug}")
            print(f"Database URL: {config.get_database_url()}")
            print(f"Redis URL: {config.get_redis_url()}")
            print(f"Log Level: {config.log_level}")
            print(f"Max Connections: {config.max_connections}")
        
        if args.export:
            with open(args.export, 'w') as f:
                yaml.dump(config.to_dict(), f, default_flow_style=False, indent=2)
            print(f"\n✓ Configuration exported to {args.export}")
        
        if args.validate:
            print("\n✓ Configuration is valid")
    
    except Exception as e:
        print(f"\n❌ Configuration error: {e}")
        return 1
    
    return 0


if __name__ == '__main__':
    exit(main())