Build a Configuration Management System

Introduction

Configuration management is essential for building flexible, maintainable applications. A good config system handles multiple environments, validates settings, supports different formats, and enables hot-reloading without restarting your application.

In this tutorial, we'll build a complete configuration management system with support for YAML, JSON, environment variables, schema validation, and more.

Core Concepts

Configuration Formats

Different formats have different strengths. We'll support:

• YAML - Human-readable, supports comments
• JSON - Standard, widely supported
• Environment Variables - For secrets and deployment

Environment-Specific Configs

Applications typically need different configs for development, staging, and production. We can handle this through file inheritance or environment overlays.

Schema Validation

Validating configuration at load time prevents runtime errors and makes debugging easier.

Project Overview

Our config system will support:

Prerequisites

• Python 3.8+ - Installed on your system
• PyYAML - Install with pip

Config Loader

Create config_manager/loader.py:

import os
import json
import yaml
from pathlib import Path
from typing import Any, Dict, Optional
from dotenv import load_dotenv


class ConfigLoader:
    def __init__(self, base_path: str = '.'):
        self.base_path = Path(base_path)
        self.config: Dict[str, Any] = {}
        self._loaded_files = []

    def load(self, 
             config_file: str = 'config.yaml',
             env: str = None,
             env_file: str = None) -> 'ConfigLoader':
        self.env = env or os.getenv('APP_ENV', 'development')
        
        if env_file:
            self._load_env_file(env_file)
        
        self._load_config_file(config_file)
        
        if self.env != 'development':
            env_config_file = f"config.{self.env}.yaml"
            if (self.base_path / env_config_file).exists():
                self._load_config_file(env_config_file)
        
        self._apply_env_overrides()
        
        return self

    def _load_config_file(self, filename: str):
        filepath = self.base_path / filename
        
        if not filepath.exists():
            return
        
        self._loaded_files.append(str(filepath))
        
        with open(filepath, 'r') as f:
            if filename.endswith('.yaml') or filename.endswith('.yml'):
                data = yaml.safe_load(f)
            elif filename.endswith('.json'):
                data = json.load(f)
            else:
                return
        
        if data:
            self._merge_config(data)

    def _load_env_file(self, filename: str):
        filepath = self.base_path / filename
        if filepath.exists():
            load_dotenv(filepath)

    def _apply_env_overrides(self):
        for key, value in os.environ.items():
            if key.startswith('APP_'):
                config_key = key[4:].lower()
                self._set_nested(config_key, value)

    def _merge_config(self, data: Dict):
        self._deep_merge(self.config, data)

    def _deep_merge(self, target: Dict, source: Dict):
        for key, value in source.items():
            if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                self._deep_merge(target[key], value)
            else:
                target[key] = value

    def _set_nested(self, key: str, value: Any):
        keys = key.split('.')
        current = self.config
        
        for k in keys[:-1]:
            if k not in current:
                current[k] = {}
            current = current[k]
        
        current[keys[-1]] = self._coerce_type(value)

    def _coerce_type(self, value: str) -> Any:
        if value.lower() == 'true':
            return True
        if value.lower() == 'false':
            return False
        if value.isdigit():
            return int(value)
        try:
            return float(value)
        except:
            return value

    def get(self, key: str, default: Any = None) -> Any:
        keys = key.split('.')
        current = self.config
        
        for k in keys:
            if isinstance(current, dict) and k in current:
                current = current[k]
            else:
                return default
        
        return current

    def set(self, key: str, value: Any):
        self._set_nested(key, value)

    def all(self) -> Dict:
        return self.config

Schema Validation

Create config_manager/validator.py:

from typing import Any, Dict, List, Callable
from dataclasses import dataclass


@dataclass
class ValidationError:
    field: str
    message: str


class SchemaValidator:
    def __init__(self):
        self.validators = {
            'string': self._validate_string,
            'integer': self._validate_integer,
            'float': self._validate_float,
            'boolean': self._validate_boolean,
            'list': self._validate_list,
            'dict': self._validate_dict,
            'enum': self._validate_enum,
            'required': self._validate_required,
        }

    def validate(self, config: Dict, schema: Dict) -> List[ValidationError]:
        errors = []
        
        for field, rules in schema.items():
            value = self._get_nested(config, field)
            
            if 'required' in rules and rules['required']:
                if value is None:
                    errors.append(ValidationError(field, 'Field is required'))
                    continue
            
            if value is None:
                continue
            
            for rule_name, rule_value in rules.items():
                validator = self.validators.get(rule_name)
                if validator:
                    error = validator(field, value, rule_value)
                    if error:
                        errors.append(error)
        
        return errors

    def _get_nested(self, config: Dict, key: str) -> Any:
        keys = key.split('.')
        current = config
        
        for k in keys:
            if isinstance(current, dict) and k in current:
                current = current[k]
            else:
                return None
        
        return current

    def _validate_string(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, str):
            return ValidationError(field, f'Expected string, got {type(value).__name__}')
        if isinstance(rule, dict):
            if 'min_length' in rule and len(value) < rule['min_length']:
                return ValidationError(field, f'String too short (min: {rule["min_length"]})')
            if 'max_length' in rule and len(value) > rule['max_length']:
                return ValidationError(field, f'String too long (max: {rule["max_length"]})')
            if 'pattern' in rule:
                import re
                if not re.match(rule['pattern'], value):
                    return ValidationError(field, f'String does not match pattern')
        return None

    def _validate_integer(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, int) or isinstance(value, bool):
            return ValidationError(field, f'Expected integer, got {type(value).__name__}')
        if isinstance(rule, dict):
            if 'min' in rule and value < rule['min']:
                return ValidationError(field, f'Value below minimum ({rule["min"]})')
            if 'max' in rule and value > rule['max']:
                return ValidationError(field, f'Value above maximum ({rule["max"]})')
        return None

    def _validate_float(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, (int, float)) or isinstance(value, bool):
            return ValidationError(field, f'Expected number, got {type(value).__name__}')
        return None

    def _validate_boolean(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, bool):
            return ValidationError(field, f'Expected boolean, got {type(value).__name__}')
        return None

    def _validate_list(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, list):
            return ValidationError(field, f'Expected list, got {type(value).__name__}')
        if isinstance(rule, dict):
            if 'min_items' in rule and len(value) < rule['min_items']:
                return ValidationError(field, f'List too short (min: {rule["min_items"]})')
            if 'max_items' in rule and len(value) > rule['max_items']:
                return ValidationError(field, f'List too long (max: {rule["max_items"]})')
        return None

    def _validate_dict(self, field: str, value: Any, rule: Any) -> ValidationError:
        if not isinstance(value, dict):
            return ValidationError(field, f'Expected object, got {type(value).__name__}')
        return None

    def _validate_enum(self, field: str, value: Any, rule: Any) -> ValidationError:
        if value not in rule:
            return ValidationError(field, f'Value must be one of {rule}')
        return None

    def _validate_required(self, field: str, value: Any, rule: Any) -> ValidationError:
        if rule and value is None:
            return ValidationError(field, 'Field is required')
        return None

Environment Handling

Create config_manager/manager.py:

import os
from typing import Any, Dict, List, Optional, Callable
from .loader import ConfigLoader
from .validator import SchemaValidator, ValidationError


class ConfigManager:
    def __init__(self, base_path: str = '.'):
        self.base_path = base_path
        self.loader = ConfigLoader(base_path)
        self.validator = SchemaValidator()
        self._callbacks: List[Callable] = []
        self._watcher = None

    def load(self, 
             config_file: str = 'config.yaml',
             env: str = None,
             env_file: str = '.env',
             schema: Dict = None) -> 'ConfigManager':
        
        self.loader.load(config_file, env, env_file)
        
        if schema:
            errors = self.validator.validate(self.loader.config, schema)
            if errors:
                raise ConfigValidationError(errors)
        
        return self

    def get(self, key: str, default: Any = None) -> Any:
        return self.loader.get(key, default)

    def set(self, key: str, value: Any):
        self.loader.set(key, value)

    def all(self) -> Dict:
        return self.loader.all()

    def on_change(self, callback: Callable):
        self._callbacks.append(callback)

    def enable_hot_reload(self, debounce: float = 1.0):
        try:
            from watchdog.observers import Observer
            from watchdog.events import FileSystemEventHandler
            
            class ConfigFileHandler(FileSystemEventHandler):
                def __init__(self, manager):
                    self.manager = manager
                    self.last_modified = 0

                def on_modified(self, event):
                    import time
                    if time.time() - self.last_modified < debounce:
                        return
                    self.last_modified = time.time()
                    
                    if event.src_path.endswith(('.yaml', '.json', '.env')):
                        self.manager.reload()
            
            self._watcher = Observer()
            handler = ConfigFileHandler(self)
            self._watcher.schedule(handler, self.base_path, recursive=True)
            self._watcher.start()
            
        except ImportError:
            print("Install watchdog for hot reload: pip install watchdog")

    def disable_hot_reload(self):
        if self._watcher:
            self._watcher.stop()
            self._watcher = None

    def reload(self):
        old_config = self.loader.config
        self.loader = ConfigLoader(self.base_path)
        
        for callback in self._callbacks:
            callback(old_config, self.loader.config)


class ConfigValidationError(Exception):
    def __init__(self, errors: List[ValidationError]):
        self.errors = errors
        message = '\n'.join(f"{e.field}: {e.message}" for e in errors)
        super().__init__(f"Configuration validation failed:\n{message}")

Hot Reload Support

Enable automatic config reloading:

from config_manager import ConfigManager

config = ConfigManager('.')

schema = {
    'database': {
        'type': 'dict',
        'schema': {
            'host': {'type': 'string', 'required': True},
            'port': {'type': 'integer', 'min': 1, 'max': 65535},
            'name': {'type': 'string', 'required': True},
        }
    },
    'server': {
        'type': 'dict',
        'schema': {
            'host': {'type': 'string'},
            'port': {'type': 'integer', 'default': 8080},
            'debug': {'type': 'boolean', 'default': False},
        }
    },
    'log_level': {
        'type': 'string',
        'enum': ['DEBUG', 'INFO', 'WARNING', 'ERROR'],
    }
}

config.load('config.yaml', schema=schema)

def on_config_change(old_config, new_config):
    print("Configuration changed!")
    print(f"Old: {old_config}")
    print(f"New: {new_config}")

config.on_change(on_config_change)
config.enable_hot_reload()

print(config.get('database.host'))
print(config.get('server.port'))

Testing the Config System

Create config files and test:

# config.yaml
app:
  name: My Application
  version: 1.0.0

database:
  host: localhost
  port: 5432
  name: myapp_dev

server:
  host: 0.0.0.0
  port: 8080
  debug: true

logging:
  level: DEBUG
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

# config.production.yaml
database:
  host: prod-db.example.com
  name: myapp_prod

server:
  port: 80
  debug: false

logging:
  level: WARNING

# .env
APP_DATABASE__HOST=override-host
APP_SERVER__DEBUG=false

# Usage
from config_manager import ConfigManager

config = ConfigManager('.')

config.load(
    config_file='config.yaml',
    env='production',
    env_file='.env'
)

print(config.get('app.name'))
print(config.get('database.host'))
print(config.get('database.port'))
print(config.get('server.debug'))

Summary

Congratulations! You've built a complete configuration management system. Here's what you learned:

• Config Loading - How to load multiple config formats
• Environment Handling - How to handle different environments
• Schema Validation - How to validate configuration
• Hot Reload - How to watch for config changes

Possible Extensions

• Add encrypted secrets support
• Implement config inheritance
• Add config versioning
• Implement remote config (etcd, Consul)