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 (equivalent to atmosphere variables, recordsdata, and dictionaries), and sizzling reloading utilizing watchdog. With a clear API and powerful validation capabilities, AsyncConfig is good 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, Non-compulsory, Kind, TypeVar, Union, get_type_hints
from dataclasses import dataclass, discipline, MISSING
from watchdog.observers import Observer
from watchdog.occasions import FileSystemEventHandler
import logging
__version__ = "0.1.0"
__author__ = "AsyncConfig Workforce"
T = TypeVar('T')
logger = logging.getLogger(__name__)
We start by importing important Python modules required for our configuration system. These embody asyncio for asynchronous operations, yaml and json for file parsing, dataclasses for structured configuration, and watchdog for warm 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."""
move
class ValidationError(ConfigError):
"""Raised when configuration validation fails."""
move
class LoadError(ConfigError):
"""Raised when configuration loading fails."""
move
@dataclass
class ConfigSource:
"""Represents a configuration supply with precedence and reload capabilities."""
path: Non-compulsory[Path] = None
env_prefix: Non-compulsory[str] = None
knowledge: Non-compulsory[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 totally different configuration-related errors, with ConfigError as the bottom class and extra particular ones, equivalent to ValidationError and LoadError, for focused troubleshooting. We additionally create a ConfigSource knowledge class to characterize a single configuration supply, which could be a file, atmosphere variables, or a dictionary, and embody assist for prioritization and non-obligatory sizzling reloading.
class ConfigWatcher(FileSystemEventHandler):
"""File system occasion handler for configuration sizzling reloading."""
def __init__(self, config_manager, paths: listing[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.data(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 sizzling reloading of configuration recordsdata. This class displays specified file paths and triggers an asynchronous reload of the configuration by the related supervisor each time a file is modified. This ensures our utility can adapt to configuration modifications in real-time with no need a restart.
class AsyncConfigManager:
"""
Trendy async configuration supervisor with sort security and sizzling reloading.
Options:
- Async-first design
- Kind-safe configuration lessons
- Setting variable assist
- Sizzling reloading
- A number of supply merging
- Validation with detailed error messages
"""
def __init__(self):
self.sources: listing[ConfigSource] = []
self.observers: listing[Observer] = []
self.config_cache: Dict[str, Any] = {}
self.reload_callbacks: listing[callable] = []
self._lock = asyncio.Lock()
def add_source(self, supply: ConfigSource) -> "AsyncConfigManager":
"""Add a configuration supply."""
self.sources.append(supply)
self.sources.kind(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 atmosphere 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: Kind[T]) -> T:
"""Load and validate configuration right into a typed dataclass."""
async with self._lock:
config_data = await self._merge_sources()
strive:
return self._validate_and_convert(config_data, config_class)
besides Exception as e:
elevate ValidationError(f"Didn't validate configuration: {e}")
async def _merge_sources(self) -> Dict[str, Any]:
"""Merge configuration from all sources based mostly on precedence."""
merged = {}
for supply in reversed(self.sources):
strive:
knowledge = await self._load_source(supply)
if knowledge:
merged.replace(knowledge)
besides Exception as e:
logger.warning(f"Didn't load supply {supply}: {e}")
return merged
async def _load_source(self, supply: ConfigSource) -> Non-compulsory[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():
elevate LoadError(f"Configuration file not discovered: {path}")
strive:
content material = await asyncio.to_thread(path.read_text)
if path.suffix.decrease() == '.json':
return json.masses(content material)
elif path.suffix.decrease() in ['.yml', '.yaml']:
return yaml.safe_load(content material) or {}
else:
elevate LoadError(f"Unsupported file format: {path.suffix}")
besides Exception as e:
elevate LoadError(f"Didn't load {path}: {e}")
def _load_env_vars(self, prefix: str) -> Dict[str, Any]:
"""Load atmosphere variables with given prefix."""
env_vars = {}
prefix = prefix.higher() + '_'
for key, worth in os.environ.objects():
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 atmosphere variable string to acceptable sort."""
if worth.decrease() in ('true', 'false'):
return worth.decrease() == 'true'
strive:
if '.' in worth:
return float(worth)
return int(worth)
besides ValueError:
move
strive:
return json.masses(worth)
besides json.JSONDecodeError:
move
return worth
def _validate_and_convert(self, knowledge: Dict[str, Any], config_class: Kind[T]) -> T:
"""Validate and convert knowledge to the desired configuration class."""
if not hasattr(config_class, '__dataclass_fields__'):
elevate ValidationError(f"{config_class.__name__} should be a dataclass")
type_hints = get_type_hints(config_class)
field_values = {}
for field_name, field_info in config_class.__dataclass_fields__.objects():
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:
elevate ValidationError(f"Required discipline '{field_name}' not present in configuration")
return config_class(**field_values)
async def start_watching(self):
"""Begin watching configuration recordsdata 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.data(f"Began watching {len(watch_paths)} configuration recordsdata")
async def stop_watching(self):
"""Cease watching configuration recordsdata."""
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."""
strive:
self.config_cache.clear()
for callback in self.reload_callbacks:
await callback()
logger.data("Configuration reloaded efficiently")
besides Exception as e:
logger.error(f"Didn't 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 (recordsdata, atmosphere variables, dictionaries), merging them by precedence, loading recordsdata asynchronously, and validating in opposition to typed dataclasses. We make the design async-first, permitting non-blocking I/O, and embody a locking mechanism to make sure protected concurrent entry. Additionally, we allow sizzling reloading by watching specified config recordsdata and triggering callbacks each time a change is detected. This setup offers a versatile, strong, and trendy basis for dynamically managing utility configurations.
async def load_config(config_class: Kind[T],
config_file: Non-compulsory[Union[str, Path]] = None,
env_prefix: Non-compulsory[str] = None,
watch: bool = False) -> T:
"""
Comfort operate to shortly load configuration.
Args:
config_class: Dataclass to load configuration into
config_file: Non-compulsory configuration file path
env_prefix: Non-compulsory atmosphere variable prefix
watch: Whether or not to observe 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 operate, load_config, to streamline the configuration setup course of. With only one name, we will load settings from a file, atmosphere variables, or each right into a typed dataclass, optionally enabling sizzling 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 utility configuration."""
debug: bool = False
log_level: str = "INFO"
secret_key: str = ""
database: DatabaseConfig = discipline(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 employees: {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)
strive:
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)
strive:
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 atmosphere, are merged based mostly on their precedence, with higher-priority sources taking priority. This highlights the pliability of managing environment-specific overrides. In demo_validation(), we validate each full and partial configurations. The system routinely fills in lacking fields with defaults the place potential. 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 capabilities."""
strive:
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__":
strive:
loop = asyncio.get_event_loop()
if loop.is_running():
print("Operating in Jupyter/IPython atmosphere")
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 capabilities, overlaying easy loading, multi-source merging, and validation. To assist each Jupyter and commonplace Python environments, we embody conditional logic for operating the demos appropriately. This ensures our configuration system is straightforward to check, showcase, and combine into a wide range of workflows proper out of the field.
In conclusion, we efficiently reveal how AsyncConfig offers a sturdy and extensible basis for managing configuration in trendy Python purposes. We see how simple 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 presents a versatile and developer-friendly strategy to handle configuration securely and effectively.
Try theΒ Full Codes.Β All credit score for this analysis goes to the researchers of this venture.
Sponsorship Alternative:Β Attain essentially the most influential AI builders in US and Europe. 1M+ month-to-month readers, 500K+ group builders, infinite potentialities.Β [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 reputation amongst audiences.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a world community of future-focused thinkers.
Unlock tomorrowβs tendencies right this moment: learn extra, subscribe to our publication, and change into a part of the NextTech group at NextTech-news.com
