On this tutorial, we information you thru the design and performance of AsyncConfig, a contemporary, async-first configuration administration library for Python. We construct it from the bottom as much as assist highly effective options, together with type-safe dataclass-based configuration loading, a number of configuration sources (reminiscent of setting variables, information, and dictionaries), and scorching reloading utilizing watchdog. With a clear API and robust validation capabilities, AsyncConfig is right for each improvement and manufacturing environments. All through this tutorial, we reveal its capabilities utilizing easy, superior, and validation-focused use circumstances, all powered by asyncio to assist non-blocking workflows.
import asyncio
import json
import os
import yaml
from pathlib import Path
from typing import Any, Dict, Optionally available, Sort, TypeVar, Union, get_type_hints
from dataclasses import dataclass, area, MISSING
from watchdog.observers import Observer
from watchdog.occasions import FileSystemEventHandler
import logging
__version__ = "0.1.0"
__author__ = "AsyncConfig Crew"
T = TypeVar('T')
logger = logging.getLogger(__name__)
We start by importing important Python modules required for our configuration system. These embrace asyncio for asynchronous operations, yaml and json for file parsing, dataclasses for structured configuration, and watchdog for decent reloading. We additionally outline some metadata and arrange a logger to trace occasions all through the system.
class ConfigError(Exception):
"""Base exception for configuration errors."""
cross
class ValidationError(ConfigError):
"""Raised when configuration validation fails."""
cross
class LoadError(ConfigError):
"""Raised when configuration loading fails."""
cross
@dataclass
class ConfigSource:
"""Represents a configuration supply with precedence and reload capabilities."""
path: Optionally available[Path] = None
env_prefix: Optionally available[str] = None
knowledge: Optionally available[Dict[str, Any]] = None
precedence: int = 0
watch: bool = False
def __post_init__(self):
if self.path:
self.path = Path(self.path)We outline a hierarchy of customized exceptions to deal with completely different configuration-related errors, with ConfigError as the bottom class and extra particular ones, reminiscent of ValidationError and LoadError, for focused troubleshooting. We additionally create a ConfigSource knowledge class to symbolize a single configuration supply, which is usually a file, setting variables, or a dictionary, and embrace assist for prioritization and optionally available scorching reloading.
class ConfigWatcher(FileSystemEventHandler):
"""File system occasion handler for configuration scorching reloading."""
def __init__(self, config_manager, paths: checklist[Path]):
self.config_manager = config_manager
self.paths = {str(p.resolve()) for p in paths}
tremendous().__init__()
def on_modified(self, occasion):
if not occasion.is_directory and occasion.src_path in self.paths:
logger.information(f"Configuration file modified: {occasion.src_path}")
asyncio.create_task(self.config_manager._reload_config())We create the ConfigWatcher class by extending FileSystemEventHandler to allow scorching reloading of configuration information. This class screens specified file paths and triggers an asynchronous reload of the configuration by the related supervisor each time a file is modified. This ensures our software can adapt to configuration modifications in real-time while not having a restart.
class AsyncConfigManager:
"""
Trendy async configuration supervisor with sort security and scorching reloading.
Options:
- Async-first design
- Sort-safe configuration courses
- Surroundings variable assist
- Scorching reloading
- A number of supply merging
- Validation with detailed error messages
"""
def __init__(self):
self.sources: checklist[ConfigSource] = []
self.observers: checklist[Observer] = []
self.config_cache: Dict[str, Any] = {}
self.reload_callbacks: checklist[callable] = []
self._lock = asyncio.Lock()
def add_source(self, supply: ConfigSource) -> "AsyncConfigManager":
"""Add a configuration supply."""
self.sources.append(supply)
self.sources.type(key=lambda x: x.precedence, reverse=True)
return self
def add_file(self, path: Union[str, Path], precedence: int = 0, watch: bool = False) -> "AsyncConfigManager":
"""Add a file-based configuration supply."""
return self.add_source(ConfigSource(path=path, precedence=precedence, watch=watch))
def add_env(self, prefix: str, precedence: int = 100) -> "AsyncConfigManager":
"""Add setting variable supply."""
return self.add_source(ConfigSource(env_prefix=prefix, precedence=precedence))
def add_dict(self, knowledge: Dict[str, Any], precedence: int = 50) -> "AsyncConfigManager":
"""Add dictionary-based configuration supply."""
return self.add_source(ConfigSource(knowledge=knowledge, precedence=precedence))
async def load_config(self, config_class: Sort[T]) -> T:
"""Load and validate configuration right into a typed dataclass."""
async with self._lock:
config_data = await self._merge_sources()
attempt:
return self._validate_and_convert(config_data, config_class)
besides Exception as e:
increase ValidationError(f"Did not validate configuration: {e}")
async def _merge_sources(self) -> Dict[str, Any]:
"""Merge configuration from all sources primarily based on precedence."""
merged = {}
for supply in reversed(self.sources):
attempt:
knowledge = await self._load_source(supply)
if knowledge:
merged.replace(knowledge)
besides Exception as e:
logger.warning(f"Did not load supply {supply}: {e}")
return merged
async def _load_source(self, supply: ConfigSource) -> Optionally available[Dict[str, Any]]:
"""Load knowledge from a single configuration supply."""
if supply.knowledge:
return supply.knowledge.copy()
if supply.path:
return await self._load_file(supply.path)
if supply.env_prefix:
return self._load_env_vars(supply.env_prefix)
return None
async def _load_file(self, path: Path) -> Dict[str, Any]:
"""Load configuration from a file."""
if not path.exists():
increase LoadError(f"Configuration file not discovered: {path}")
attempt:
content material = await asyncio.to_thread(path.read_text)
if path.suffix.decrease() == '.json':
return json.hundreds(content material)
elif path.suffix.decrease() in ['.yml', '.yaml']:
return yaml.safe_load(content material) or {}
else:
increase LoadError(f"Unsupported file format: {path.suffix}")
besides Exception as e:
increase LoadError(f"Did not load {path}: {e}")
def _load_env_vars(self, prefix: str) -> Dict[str, Any]:
"""Load setting variables with given prefix."""
env_vars = {}
prefix = prefix.higher() + '_'
for key, worth in os.environ.gadgets():
if key.startswith(prefix):
config_key = key[len(prefix):].decrease()
env_vars[config_key] = self._convert_env_value(worth)
return env_vars
def _convert_env_value(self, worth: str) -> Any:
"""Convert setting variable string to applicable sort."""
if worth.decrease() in ('true', 'false'):
return worth.decrease() == 'true'
attempt:
if '.' in worth:
return float(worth)
return int(worth)
besides ValueError:
cross
attempt:
return json.hundreds(worth)
besides json.JSONDecodeError:
cross
return worth
def _validate_and_convert(self, knowledge: Dict[str, Any], config_class: Sort[T]) -> T:
"""Validate and convert knowledge to the desired configuration class."""
if not hasattr(config_class, '__dataclass_fields__'):
increase ValidationError(f"{config_class.__name__} have to be a dataclass")
type_hints = get_type_hints(config_class)
field_values = {}
for field_name, field_info in config_class.__dataclass_fields__.gadgets():
if field_name in knowledge:
field_value = knowledge[field_name]
if hasattr(field_info.sort, '__dataclass_fields__'):
if isinstance(field_value, dict):
field_value = self._validate_and_convert(field_value, field_info.sort)
field_values[field_name] = field_value
elif field_info.default just isn't MISSING:
field_values[field_name] = field_info.default
elif field_info.default_factory just isn't MISSING:
field_values[field_name] = field_info.default_factory()
else:
increase ValidationError(f"Required area '{field_name}' not present in configuration")
return config_class(**field_values)
async def start_watching(self):
"""Begin watching configuration information for modifications."""
watch_paths = []
for supply in self.sources:
if supply.watch and supply.path:
watch_paths.append(supply.path)
if watch_paths:
observer = Observer()
watcher = ConfigWatcher(self, watch_paths)
for path in watch_paths:
observer.schedule(watcher, str(path.guardian), recursive=False)
observer.begin()
self.observers.append(observer)
logger.information(f"Began watching {len(watch_paths)} configuration information")
async def stop_watching(self):
"""Cease watching configuration information."""
for observer in self.observers:
observer.cease()
observer.be a part of()
self.observers.clear()
async def _reload_config(self):
"""Reload configuration from all sources."""
attempt:
self.config_cache.clear()
for callback in self.reload_callbacks:
await callback()
logger.information("Configuration reloaded efficiently")
besides Exception as e:
logger.error(f"Did not reload configuration: {e}")
def on_reload(self, callback: callable):
"""Register a callback to be known as when configuration is reloaded."""
self.reload_callbacks.append(callback)
async def __aenter__(self):
await self.start_watching()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.stop_watching()We now implement the core of our system by the AsyncConfigManager class. It acts because the central controller for all configuration operations, including sources (information, setting variables, dictionaries), merging them by precedence, loading information asynchronously, and validating in opposition to typed dataclasses. We make the design async-first, permitting non-blocking I/O, and embrace a locking mechanism to make sure protected concurrent entry. Additionally, we allow scorching reloading by watching specified config information and triggering callbacks each time a change is detected. This setup offers a versatile, strong, and fashionable basis for dynamically managing software configurations.
async def load_config(config_class: Sort[T],
config_file: Optionally available[Union[str, Path]] = None,
env_prefix: Optionally available[str] = None,
watch: bool = False) -> T:
"""
Comfort perform to shortly load configuration.
Args:
config_class: Dataclass to load configuration into
config_file: Optionally available configuration file path
env_prefix: Optionally available setting variable prefix
watch: Whether or not to look at for file modifications
Returns:
Configured occasion of config_class
"""
supervisor = AsyncConfigManager()
if config_file:
supervisor.add_file(config_file, precedence=0, watch=watch)
if env_prefix:
supervisor.add_env(env_prefix, precedence=100)
return await supervisor.load_config(config_class)We add a handy helper perform, load_config, to streamline the configuration setup course of. With only one name, we will load settings from a file, setting variables, or each right into a typed dataclass, optionally enabling scorching reloading. This utility makes the library beginner-friendly whereas nonetheless supporting superior use circumstances beneath the hood.
@dataclass
class DatabaseConfig:
"""Instance database configuration."""
host: str = "localhost"
port: int = 5432
username: str = "admin"
password: str = ""
database: str = "myapp"
ssl_enabled: bool = False
pool_size: int = 10
@dataclass
class AppConfig:
"""Instance software configuration."""
debug: bool = False
log_level: str = "INFO"
secret_key: str = ""
database: DatabaseConfig = area(default_factory=DatabaseConfig)
redis_url: str = "redis://localhost:6379"
max_workers: int = 4
async def demo_simple_config():
"""Demo easy configuration loading."""
sample_config = {
"debug": True,
"log_level": "DEBUG",
"secret_key": "dev-secret-key",
"database": {
"host": "localhost",
"port": 5432,
"username": "testuser",
"password": "testpass",
"database": "testdb"
},
"max_workers": 8
}
supervisor = AsyncConfigManager()
supervisor.add_dict(sample_config, precedence=0)
config = await supervisor.load_config(AppConfig)
print("=== Easy Configuration Demo ===")
print(f"Debug mode: {config.debug}")
print(f"Log stage: {config.log_level}")
print(f"Database host: {config.database.host}")
print(f"Database port: {config.database.port}")
print(f"Max staff: {config.max_workers}")
return configWe outline two instance configuration dataclasses: DatabaseConfig and AppConfig, which showcase how nested and typed configurations are structured. To reveal actual utilization, we write demo_simple_config(), the place we load a primary dictionary into our config supervisor. This illustrates how effortlessly we will map structured knowledge into type-safe Python objects, making configuration dealing with clear, readable, and maintainable.
async def demo_advanced_config():
"""Demo superior configuration with a number of sources."""
base_config = {
"debug": False,
"log_level": "INFO",
"secret_key": "production-secret",
"max_workers": 4
}
override_config = {
"debug": True,
"log_level": "DEBUG",
"database": {
"host": "dev-db.instance.com",
"port": 5433
}
}
env_config = {
"secret_key": "env-secret-key",
"redis_url": "redis://prod-redis:6379"
}
print("n=== Superior Configuration Demo ===")
supervisor = AsyncConfigManager()
supervisor.add_dict(base_config, precedence=0)
supervisor.add_dict(override_config, precedence=50)
supervisor.add_dict(env_config, precedence=100)
config = await supervisor.load_config(AppConfig)
print("Configuration sources merged:")
print(f"Debug mode: {config.debug} (from override)")
print(f"Log stage: {config.log_level} (from override)")
print(f"Secret key: {config.secret_key} (from env)")
print(f"Database host: {config.database.host} (from override)")
print(f"Redis URL: {config.redis_url} (from env)")
return config
async def demo_validation():
"""Demo configuration validation."""
print("n=== Configuration Validation Demo ===")
valid_config = {
"debug": True,
"log_level": "DEBUG",
"secret_key": "test-key",
"database": {
"host": "localhost",
"port": 5432
}
}
supervisor = AsyncConfigManager()
supervisor.add_dict(valid_config, precedence=0)
attempt:
config = await supervisor.load_config(AppConfig)
print("✓ Legitimate configuration loaded efficiently")
print(f" Database SSL: {config.database.ssl_enabled} (default worth)")
print(f" Database pool dimension: {config.database.pool_size} (default worth)")
besides ValidationError as e:
print(f"✗ Validation error: {e}")
incomplete_config = {
"debug": True,
"log_level": "DEBUG"
}
manager2 = AsyncConfigManager()
manager2.add_dict(incomplete_config, precedence=0)
attempt:
config2 = await manager2.load_config(AppConfig)
print("✓ Configuration with defaults loaded efficiently")
print(f" Secret key: '{config2.secret_key}' (default empty string)")
besides ValidationError as e:
print(f"✗ Validation error: {e}")We reveal superior options of our config system by two examples. In demo_advanced_config(), we reveal how a number of configuration sources, base, override, and setting, are merged primarily based on their precedence, with higher-priority sources taking priority. This highlights the flexibleness of managing environment-specific overrides. In demo_validation(), we validate each full and partial configurations. The system mechanically fills in lacking fields with defaults the place doable. It throws clear ValidationErrors when required fields are lacking, making certain sort security and strong configuration administration in real-world purposes.
async def run_demos():
"""Run all demonstration features."""
attempt:
await demo_simple_config()
await demo_advanced_config()
await demo_validation()
print("n=== All demos accomplished efficiently! ===")
besides Exception as e:
print(f"Demo error: {e}")
import traceback
traceback.print_exc()
await run_demos()
if __name__ == "__main__":
attempt:
loop = asyncio.get_event_loop()
if loop.is_running():
print("Operating in Jupyter/IPython setting")
print("Use: await run_demos()")
else:
asyncio.run(run_demos())
besides RuntimeError:
asyncio.run(run_demos())
We conclude the tutorial with run_demos(), a utility that sequentially executes all demonstration features, overlaying easy loading, multi-source merging, and validation. To assist each Jupyter and customary Python environments, we embrace conditional logic for operating the demos appropriately. This ensures our configuration system is straightforward to check, showcase, and combine into quite a lot of workflows proper out of the field.
In conclusion, we efficiently reveal how AsyncConfig offers a strong and extensible basis for managing configuration in fashionable Python purposes. We see how straightforward it’s to merge a number of sources, validate configurations in opposition to typed schemas, and reply to reside file modifications in real-time. Whether or not we’re constructing microservices, async backends, or CLI instruments, this library gives a versatile and developer-friendly approach to handle configuration securely and effectively.
Try the Full Codes. All credit score for this analysis goes to the researchers of this challenge.
Sponsorship Alternative: Attain probably the most influential AI builders in US and Europe. 1M+ month-to-month readers, 500K+ group builders, infinite prospects. [Explore Sponsorship]
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its recognition amongst audiences.
