Async Configuration

Async/await support for Confii.

This module provides async versions of Config and loaders for use in asynchronous Python applications.

class confii.async_config.AsyncConfig(env: str | None = None, loaders: Sequence[AsyncLoader] | None = None, use_env_expander: bool = True, use_type_casting: bool = True, debug_mode: bool = False, deep_merge: bool = True, secret_resolver: SecretResolver | None = None)

Bases: object

Async version of Config for use in async applications.

This class provides async methods for loading and managing configuration in asynchronous Python applications.

Example

>>> async def main():
...     loader = AsyncYamlLoader("config.yaml")
...     config = await AsyncConfig.create(loaders=[loader])
...     value = await config.get_async("database.host")

__init__(env: str | None = None, loaders: Sequence[AsyncLoader] | None = None, use_env_expander: bool = True, use_type_casting: bool = True, debug_mode: bool = False, deep_merge: bool = True, secret_resolver: SecretResolver | None = None) → None

Initialize AsyncConfig instance.

Parameters:

• env – Environment name

• loaders – List of AsyncLoader instances

• use_env_expander – Enable environment variable expansion

• use_type_casting – Enable automatic type casting

• debug_mode – Enable debug mode

• deep_merge – Enable deep merging

• secret_resolver – Optional secret resolver

async classmethod create(env: str | None = None, loaders: Sequence[AsyncLoader] | None = None, **kwargs: Any) → AsyncConfig

Create and initialize AsyncConfig asynchronously.

This is the preferred way to create an AsyncConfig instance as it properly handles async initialization.

Parameters:

• env – Environment name

• loaders – List of AsyncLoader instances

• **kwargs – Additional arguments passed to __init__

Returns:

Initialized AsyncConfig instance

Example

>>> config = await AsyncConfig.create(
...     env="production", loaders=[AsyncYamlLoader("config.yaml")]
... )

async get_async(key_path: str, default: Any = None) → Any

Get configuration value asynchronously.

Parameters:

• key_path – Dot-separated key path

• default – Default value if key not found

Returns:

Configuration value or default

async load() → None

Load configuration from all async loaders.

This method must be called before accessing configuration values.

Raises:

ConfigLoadError – If loading fails

async reload() → None

Reload configuration asynchronously.

This method reloads all configurations from their sources.

to_dict() → Dict[str, Any]

Export configuration as dictionary.

Returns:

Configuration dictionary

async validate_async(schema: Any | None = None) → bool

Validate configuration asynchronously.

Parameters:

schema – Optional schema to validate against

Returns:

True if valid, False otherwise

class confii.async_config.AsyncHTTPLoader(url: str, timeout: int = 30)

Bases: AsyncLoader

Async HTTP configuration loader.

Fetches configuration from a remote HTTP/HTTPS endpoint using aiohttp. The response format (JSON or YAML) is auto-detected from the Content-Type header or the URL file extension.

Requires the aiohttp package (pip install aiohttp).

Example

>>> loader = AsyncHTTPLoader("https://cfg.example.com/app.json", timeout=10)
>>> config = await loader.load()

__init__(url: str, timeout: int = 30) → None

Initialize async HTTP loader.

Parameters:

• url – HTTP/HTTPS URL to load configuration from

• timeout – Request timeout in seconds

async load() → Dict[str, Any]

Load configuration from HTTP endpoint (async).

Returns:

Loaded configuration dictionary

Raises:

ConfigLoadError – If loading fails

class confii.async_config.AsyncLoader(source: str)

Bases: object

Abstract base class for async configuration loaders.

All async loaders must inherit from this class and implement the load() method as a coroutine. Use async loaders when your configuration sources involve I/O that benefits from non-blocking execution (remote APIs, cloud storage, etc.).

Example

>>> class AsyncRedisLoader(AsyncLoader):
...     async def load(self):
...         import aioredis
...
...         r = await aioredis.from_url(self.source)
...         return await r.hgetall("config")

__init__(source: str) → None

Initialize the async loader.

Parameters:

source – The source identifier (file path, URL, prefix, etc.)

async load() → Dict[str, Any] | None

Load configuration from the source (async).

This method must be implemented by subclasses.

Returns:

Dictionary containing the loaded configuration, or None if the source doesn’t exist or couldn’t be loaded.

Raises:

ConfigLoadError – If loading fails due to an error

class confii.async_config.AsyncYamlLoader(source: str)

Bases: AsyncLoader

Async YAML configuration loader.

Reads a YAML file using a thread-pool executor so the event loop is never blocked by disk I/O. Suitable for asyncio-based applications that load configuration at startup.

Example

>>> loader = AsyncYamlLoader("config.yaml")
>>> config = await loader.load()
>>> print(config["database"]["host"])

async load() → Dict[str, Any] | None

Load configuration from YAML file (async).

Returns:

Loaded configuration dictionary, or None if file doesn’t exist

Raises:

ConfigLoadError – If loading fails