API Reference

Manage configuration with pydantic.

Modules:

• config –

  The base configuration model class BaseConfig.

• context –

  Facilities for contextual processing.

• contrib –

  configzen.contrib: miscellaneous reusable configuration models from configzen.

• data –

  Interfaces for encapsulation of configuring and using data formats.

• errors –

  Specialized exceptions raised by configzen.

• formats –

  configzen.formats: Data formats supported natively by configzen.

• module_proxy –

  Runtime modules with attribute type validation.

• processor –

  Replacement API processor for configuration data.

• routes –

  Routes creation and parsing.

• sources –

  Sources and destinations that hold the configuration data.

• typedefs –

  Miscellaneous type definitions for configzen.

config

The base configuration model class BaseConfig.

Classes:

• ModelConfig –

  Meta-configuration for configzen models.

• BaseConfig –

  Base class for all configuration models.

Functions:

• config_step –

  Return the value of a configuration attribute.

ModelConfig

Bases: SettingsConfigDict

Meta-configuration for configzen models.

BaseConfig(**data)

Bases: BaseSettings

Base class for all configuration models.

Methods:

• config_find_routes –

  Locate all occurrences of a subconfiguration in the current configuration.

• config_find_route –

  Locate exactly one (closest) route to the given subconfiguration.

• config_load –

  Load this configuration from a given source.

• config_load_async –

  Do the same as config_load, but asynchronously (no I/O blocking).

• config_reload –

  Reload the configuration from the same source.

• config_reload_async –

  Do the same as config_reload asynchronously (no I/O blocking).

• config_save –

  Save the configuration to a given destination.

• config_save_async –

  Do the same as config_save, but asynchronously (no I/O blocking).

• config_at –

  Return a configuration item at the given set of routes.

• config_dump –

  Return a dictionary representation of the configuration.

• __getitem__ –

  Return a configuration item at the given set of routes.

• __setitem__ –

  Set a configuration item at the given set of routes.

• __init_subclass__ –

  Initialize the configuration subclass.

Attributes:

• config_root (BaseConfig) –

  Return the root configuration that was used to load the entire data.

• config_source (ConfigSource[Any, Any] | None) –

  Return the configuration source that was used to load the configuration.

• config_data (Data) –

  Return the configuration that was loaded from the configuration source.

• config_processor (ConfigProcessor) –

  Current configuration processor.

Source code in configzen/config.py

def __init__(self, **data: Any) -> None:
    try:
        owner = owner_lookup.get()
    except LookupError:
        owner = None
        if processing.get():
            owner_lookup.set(self)
    super().__init__(**data)
    self._config_root = owner

config_root: BaseConfig

Return the root configuration that was used to load the entire data.

config_source: ConfigSource[Any, Any] | None

Return the configuration source that was used to load the configuration.

config_data: Data

Return the configuration that was loaded from the configuration source.

config_processor: ConfigProcessor

Current configuration processor.

Processor stores the initial data used when loading the configuration, resolves macros etc.

config_find_routes(subconfig)

Locate all occurrences of a subconfiguration in the current configuration.

Return a set of routes to the located subconfiguration.

Source code in configzen/config.py

def config_find_routes(
    self,
    subconfig: BaseConfig,
) -> set[Route]:
    """
    Locate all occurrences of a subconfiguration in the current configuration.

    Return a set of routes to the located subconfiguration.
    """
    if not isinstance(subconfig, BaseConfig):
        msg = f"Expected a BaseConfig subclass instance, got {type(subconfig)!r}"
        raise TypeError(msg)
    return set(
        _locate_in_mapping(vars(self), subconfig, attribute_access=True),
    )

config_find_route(subconfig)

Locate exactly one (closest) route to the given subconfiguration.

Source code in configzen/config.py

def config_find_route(self, subconfig: BaseConfig) -> Route:
    """Locate exactly one (closest) route to the given subconfiguration."""
    all_routes = self.config_find_routes(subconfig)
    if not all_routes:
        msg = f"Unable to locate subconfiguration {subconfig}"
        raise LookupError(msg)
    return next(iter(all_routes))

config_load(source=None, *, processor_factory=None)

Load this configuration from a given source.

Parameters:

• source

  (object | None, default: None ) –

  Where to load the configuration from. The argument passed is forwarded to confizen.sources.get_config_source() which will resolve the intended configuration source: for example, "abc.ini" will be resolved to a TOML text file source. Keep in mind, however, that for binary formats such as non-XML Plist you must specify its format type to binary, so in that case just create BinaryFileConfigSource("plist_file.plist").

• processor_factory

  (Callable[..., ConfigProcessor] | None, default: None ) –

  The state factory to use to parse the newly loaded configuration data.

Returns:

• self –

Source code in configzen/config.py

@classmethod
@isolated_context_function
def config_load(
    cls,
    source: object | None = None,
    *,
    processor_factory: Callable[..., ConfigProcessor] | None = None,
) -> Self:
    """
    Load this configuration from a given source.

    Parameters
    ----------
    source
        Where to load the configuration from. The argument passed is forwarded
        to `confizen.sources.get_config_source()` which will resolve
        the intended configuration source: for example, "abc.ini" will be resolved
        to a TOML text file source. Keep in mind, however, that for binary formats
        such as non-XML Plist you must specify its format type to binary, so in
        that case just create `BinaryFileConfigSource("plist_file.plist")`.
    processor_factory
        The state factory to use to parse the newly loaded configuration data.

    Returns
    -------
    self

    """
    cls._try_rebuild_model()

    # Validate the source we load our configuration from.
    config_source = cls._validate_config_source(source)

    # Validate the processor we use to parse the loaded configuration data.
    make_processor = cls._validate_processor_factory(processor_factory)

    # Load the configuration data from the sanitized source.
    # Keep in mind the loaded data object keeps all the additional
    # metadata that we want to keep.
    # Then we pass it to the processor factory to process the configuration data
    # into a bare dictionary that does not hold anything else
    # than the configuration data, by using `processor.get_processed_data()`.
    processor = make_processor(config_source.load())

    processing_context = ProcessingContext(cls, processor, trace=[config_source])
    processing.set(processing_context)
    # ruff: noqa: FBT003
    try:
        # Processing will execute any commands that are present
        # in the configuration data and return the final configuration
        # data that we will use to construct an instance of the configuration model.
        # During this process, we lose all the additional metadata that we
        # want to keep in the configuration data.
        # They will be added back to the exported data when the configuration
        # is saved (`processor.revert_processor_changes()`).
        self = cls(**processor.get_processed_data())
    finally:
        processing.set(None)

    # Quick setup and we're done.
    self._config_source = config_source
    self._config_processor = processor
    return self

config_load_async(source=None, *, processor_factory=None)

Do the same as config_load, but asynchronously (no I/O blocking).

Parameters:

• source

  (object | None, default: None ) –

  Where to load the configuration from. The argument passed is forwarded to confizen.sources.get_config_source() which will resolve the intended configuration source: for example, "abc.ini" will be resolved to a TOML text file source. Keep in mind, however, that for binary formats such as non-XML Plist you must specify its format type to binary, so in that case just create BinaryFileConfig"plist_file.plist").

• processor_factory

  (Callable[..., ConfigProcessor] | None, default: None ) –

  The state factory to use to parse the newly loaded configuration data.

Returns:

• self –

Source code in configzen/config.py

@classmethod
@isolated_context_coroutine
async def config_load_async(
    cls,
    source: object | None = None,
    *,
    processor_factory: Callable[..., ConfigProcessor] | None = None,
) -> Self:
    """
    Do the same as `config_load`, but asynchronously (no I/O blocking).

    Parameters
    ----------
    source
        Where to load the configuration from. The argument passed is forwarded
        to `confizen.sources.get_config_source()` which will resolve
        the intended configuration source: for example, "abc.ini" will be resolved
        to a TOML text file source. Keep in mind, however, that for binary formats
        such as non-XML Plist you must specify its format type to binary, so in
        that case just create `BinaryFileConfig"plist_file.plist")`.
    processor_factory
        The state factory to use to parse the newly loaded configuration data.

    Returns
    -------
    self

    """
    cls._try_rebuild_model()

    # Intentionally not using `run_sync(config_load)` here.
    # We want to keep make the set up instructions blocking to avoid running
    # into mutexes.

    config_source = cls._validate_config_source(source)
    make_processor = cls._validate_processor_factory(processor_factory)
    processor = make_processor(await config_source.load_async())

    processing_context = ProcessingContext(cls, processor, trace=[config_source])
    processing.set(processing_context)
    try:
        self = cls(**await run_sync(processor.get_processed_data))
    finally:
        processing.set(None)

    self._config_processor = processor
    self._config_source = config_source
    return self

config_reload()

Reload the configuration from the same source.

Source code in configzen/config.py

def config_reload(self) -> Self:
    """Reload the configuration from the same source."""
    source = self.config_source

    if source is None:
        msg = "Cannot reload a manually instantiated configuration"
        raise RuntimeError(msg)

    root = self.config_root

    # Create a new processor with the same options as the current one.
    processor = root.config_processor.create_processor(source.load())

    # Construct a new configuration instance.
    # Respect __class__ attribute in case root might be a proxy (from proxyvars).
    new_root = root.__class__(**processor.get_processed_data())

    # Copy values from the freshly loaded configuration into our instance.
    if root is self:
        new_data = new_root.config_dump()
    else:
        route_to_self = root.config_find_route(self)
        new_data = cast("Self", route_to_self.get(new_root)).config_dump()

    for key, value in new_data.items():
        setattr(self, key, value)

    return self

config_reload_async()

Do the same as config_reload asynchronously (no I/O blocking).

Source code in configzen/config.py

async def config_reload_async(self) -> Self:
    """Do the same as `config_reload` asynchronously (no I/O blocking)."""
    source = self.config_source

    if source is None:
        msg = "Cannot reload a manually instantiated configuration"
        raise RuntimeError(msg)

    root = self.config_root

    # Create a new state processor the same options as the current one.
    processor = root.config_processor.create_processor(source.load())

    # Construct a new configuration instance.
    new_root = root.__class__(**await run_sync(processor.get_processed_data))

    # Copy values from the freshly loaded configuration into our instance.
    if root is self:
        new_data = new_root.config_dump()
    else:
        route_to_self = root.config_find_route(self)
        new_data = cast("Self", route_to_self.get(new_root)).config_dump()

    for key, value in new_data.items():
        setattr(self, key, value)

    return self

config_save(destination=None)

Save the configuration to a given destination.

Parameters:

• destination

  (object | None, default: None ) –

  Where to save the configuration to. The argument passed is forwarded to confizen.sources.get_config_source() which will resolve the intended configuration source: for example, "abc.ini" will be resolved to a TOML text file source. Keep in mind, however, that for binary formats such as non-XML Plist you must specify its format type to binary, so in that case just create BinaryFileConfigSource("plist_file.plist").

Source code in configzen/config.py

def config_save(self, destination: object | None = None) -> Self:
    """
    Save the configuration to a given destination.

    Parameters
    ----------
    destination
        Where to save the configuration to. The argument passed is forwarded
        to `confizen.sources.get_config_source()` which will resolve
        the intended configuration source: for example, "abc.ini" will be resolved
        to a TOML text file source. Keep in mind, however, that for binary formats
        such as non-XML Plist you must specify its format type to binary, so in
        that case just create `BinaryFileConfigSource("plist_file.plist")`.

    """
    config_destination, data = self._config_data_save(destination)
    config_destination.dump(data)
    return self

config_save_async(destination=None)

Do the same as config_save, but asynchronously (no I/O blocking).

Parameters:

• destination

  (object | None, default: None ) –

  Where to save the configuration to. The argument passed is forwarded to confizen.sources.get_config_source() which will resolve the intended configuration source: for example, "abc.ini" will be resolved to a TOML text file source. Keep in mind, however, that for binary formats such as non-XML Plist you must specify its format type to binary, so in that case just create BinaryFileConfigSource("plist_file.plist").

Source code in configzen/config.py

async def config_save_async(self, destination: object | None = None) -> Self:
    """
    Do the same as `config_save`, but asynchronously (no I/O blocking).

    Parameters
    ----------
    destination
        Where to save the configuration to. The argument passed is forwarded
        to `confizen.sources.get_config_source()` which will resolve
        the intended configuration source: for example, "abc.ini" will be resolved
        to a TOML text file source. Keep in mind, however, that for binary formats
        such as non-XML Plist you must specify its format type to binary, so in
        that case just create `BinaryFileConfigSource("plist_file.plist")`.

    """
    config_destination, data = self._config_data_save(destination)
    await config_destination.dump_async(data)
    return self

config_at(*routes)

Return a configuration item at the given set of routes.

Source code in configzen/config.py

def config_at(self, *routes: RouteLike) -> Item:
    """Return a configuration item at the given set of routes."""
    return Item(routes=set(map(Route, routes)), config=self)

config_dump()

Return a dictionary representation of the configuration.

Source code in configzen/config.py

def config_dump(self) -> dict[str, object]:
    """Return a dictionary representation of the configuration."""
    return super().model_dump()

__getitem__(routes)

Return a configuration item at the given set of routes.

Source code in configzen/config.py

def __getitem__(self, routes: RouteLike | tuple[RouteLike, ...]) -> Item:
    """Return a configuration item at the given set of routes."""
    if isinstance(routes, tuple):
        return self.config_at(*routes)
    return self.config_at(routes)

__setitem__(item, value)

Set a configuration item at the given set of routes.

Source code in configzen/config.py

def __setitem__(self, item: RouteLike, value: Any) -> None:
    """Set a configuration item at the given set of routes."""
    self.config_at(item).config = value

__init_subclass__(**kwargs)

Initialize the configuration subclass.

Source code in configzen/config.py

def __init_subclass__(cls, **kwargs: Unpack[ModelConfig]) -> None:
    """Initialize the configuration subclass."""
    super().__init_subclass__(**cast("BaseConfigDict", kwargs))

config_step(owner, _annotation, step)

Return the value of a configuration attribute.

Source code in configzen/config.py

@advance_linked_route.register(BaseConfig)
def config_step(
    owner: type[BaseConfig],
    _annotation: Any,
    step: Step[Any],
) -> Any:
    """Return the value of a configuration attribute."""
    return owner.model_fields[step.key].annotation

context

Facilities for contextual processing.

Functions:

• isolated_context_function –

  Copy the context automatically on function call.

• isolated_context_coroutine –

  Copy the context automatically on coroutine execution.

• run_isolated –

  Run a function in an isolated context.

• async_run_isolated –

  Await a coroutine in an isolated context.

isolated_context_function(func)

Copy the context automatically on function call.

Allows to isolate the library context from the user context.

Used as a decorator.

Source code in configzen/context.py

def isolated_context_function(func: Callable[_P, _T]) -> Callable[_P, _T]:
    """
    Copy the context automatically on function call.

    Allows to isolate the library context from the user context.

    Used as a decorator.
    """
    if isinstance(func, (classmethod, staticmethod)):
        return type(func)(isolated_context_function(func.__func__))

    @wraps(func)
    def copy(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        return run_isolated(func, *args, **kwargs)

    return copy

isolated_context_coroutine(func)

Copy the context automatically on coroutine execution.

Allows to isolate library context from the user context.

Used as a decorator.

Source code in configzen/context.py

def isolated_context_coroutine(
    func: Callable[_P, Coroutine[object, object, _T]],
) -> Callable[_P, Coroutine[object, object, _T]]:
    """
    Copy the context automatically on coroutine execution.

    Allows to isolate library context from the user context.

    Used as a decorator.
    """
    if isinstance(func, (classmethod, staticmethod)):
        return type(func)(isolated_context_coroutine(func.__func__))

    @wraps(func)
    async def copy_async(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        return await async_run_isolated(func, *args, **kwargs)

    return copy_async

run_isolated(func, *args, **kwargs)

Run a function in an isolated context.

Source code in configzen/context.py

def run_isolated(func: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs) -> _T:
    """Run a function in an isolated context."""
    context = contextvars.copy_context()
    return context.run(func, *args, **kwargs)

async_run_isolated(func, *args, **kwargs)

Await a coroutine in an isolated context.

Source code in configzen/context.py

def async_run_isolated(
    func: Callable[_P, Coroutine[object, object, _T]],
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> asyncio.Task[_T]:
    """Await a coroutine in an isolated context."""
    return asyncio.create_task(func(*args, **kwargs))

contrib

configzen.contrib: miscellaneous reusable configuration models from configzen.

Modules:

• pyproject –

  Configuration model for pyproject.toml files.

• setup –

  Configuration model for pyproject.toml files.

pyproject

Configuration model for pyproject.toml files.

setup

Configuration model for pyproject.toml files.

data

Interfaces for encapsulation of configuring and using data formats.

Classes:

• DataFormatOptions –

  Base class for indicating possible options to configure a data format.

• DataFormat –

  Core interface for configuring and using any data format through within configzen.

Functions:

• roundtrip_update_mapping –

  Update a mapping without losing recursively attached metadata.

• roundtrip_update_sequence –

  Update a sequence without losing recursively attached metadata.

Attributes:

• BinaryDataFormat –

  Core interface for configuring and using binary data formats through

• TextDataFormat –

  Core interface for configuring and using text data formats through

BinaryDataFormat = DataFormat[DataFormatOptionsType, bytes]

Core interface for configuring and using binary data formats through within configzen.

Do not use this class directly. If you need to implement your own binary data format, implement a subclass of this class. Remember to ensure that your subclass is executed, so that it gets registered to the registry of data formats.

TextDataFormat = DataFormat[DataFormatOptionsType, str]

Core interface for configuring and using text data formats through within configzen.

Do not use this class directly. If you need to implement your own text data format, implement a subclass of this class. Remember to ensure that your subclass is executed, so that it gets registered to the registry of data formats.

DataFormatOptions

Bases: TypedDict

Base class for indicating possible options to configure a data format.

DataFormat(options=None)

Bases: Generic[DataFormatOptionsType, AnyStr]

Core interface for configuring and using any data format through within configzen.

Do not use this class directly. If you need to implement your own data format, implement a subclass of: - BinaryDataFormat, if it is a bitwise data format, or - TextDataFormat, if it is a text data format.

Methods:

• for_extension –

  Create a data format instance for an extension.

• is_binary –

  Return whether the data format is bitwise.

• configure –

  Configure the data format.

• load –

  Load the data from a stream.

• dump –

  Load the data from a stream.

• register_file_extensions –

  Register the file extensions supported by this data format.

• validate_source –

  Validate the config source.

• roundtrip_update_mapping –

  Update the loaded data in a round-trip manner.

• roundtrip_update_sequence –

  Merge new data sequence without losing comments.

• __init_subclass__ –

  Subclass hook. Pass skip_hook=True to skip it.

Source code in configzen/data.py

def __init__(self, options: DataFormatOptionsType | None = None) -> None:
    self.configure(**(options or {}))

for_extension(extension_name, options=None)

Create a data format instance for an extension.

Source code in configzen/data.py

@classmethod
def for_extension(
    cls,
    extension_name: str,
    options: DataFormatOptionsType | None = None,
) -> DataFormat[Any, Any]:
    """Create a data format instance for an extension."""
    return cls.extension_registry[extension_name](options)

is_binary()

is_binary() -> Literal[True]

is_binary() -> Literal[False]

Return whether the data format is bitwise.

Source code in configzen/data.py

def is_binary(self) -> bool:
    """Return whether the data format is bitwise."""
    return type_check(self, DataFormat[Any, bytes])

configure(**options)

Configure the data format.

Every data format provides its own options, related to comments, indentation, and other format-specific features.

Source code in configzen/data.py

def configure(self, **options: Unpack[DataFormatOptions]) -> None:
    """
    Configure the data format.

    Every data format provides its own options, related to comments, indentation,
    and other format-specific features.
    """

load(stream)

Load the data from a stream.

Return a mutable mapping representing the loaded data which is mutation-sensitive (for round-trip processing).

Every configuration source transforms the input data into a stream to be processed by the data format, because most data format libraries operate on streams.

This method is called by the config instance.

Source code in configzen/data.py

@abstractmethod
def load(self, stream: IO[AnyStr]) -> Data:
    """
    Load the data from a stream.

    Return a mutable mapping representing the loaded data
    which is mutation-sensitive (for round-trip processing).

    Every configuration source transforms the input data into a stream
    to be processed by the data format, because most data format libraries
    operate on streams.

    This method is called by the config instance.
    """

dump(data, stream)

Load the data from a stream.

Every configuration source transforms the input data into a stream to be processed by the data format, because most libraries implementing data formats operate on streams.

This method is called by the config instance.

Source code in configzen/data.py

@abstractmethod
def dump(self, data: Data, stream: IO[AnyStr]) -> None:
    """
    Load the data from a stream.

    Every configuration source transforms the input data into a stream
    to be processed by the data format, because most libraries implementing
    data formats operate on streams.

    This method is called by the config instance.
    """

register_file_extensions()

Register the file extensions supported by this data format.

Source code in configzen/data.py

@classmethod
def register_file_extensions(cls) -> None:
    """Register the file extensions supported by this data format."""
    cls.extension_registry.update(dict.fromkeys(cls.file_extensions, cls))

validate_source(source)

Validate the config source.

Source code in configzen/data.py

def validate_source(self, source: ConfigSource[Any, AnyStr]) -> None:
    """Validate the config source."""
    if self.is_binary() and not source.is_binary():
        msg = (
            f"{source} is not a binary source, "
            f"but {self.__class__.__name__} is a binary data format"
        )
        raise TypeError(msg)

roundtrip_update_mapping(roundtrip_data, mergeable_data)

Update the loaded data in a round-trip manner.

Use values from the configuration altered programmatically in runtime, while keeping the structure and comments of the original data.

Parameters:

• roundtrip_data

  (Data) –

  The data to be updated. Stores the original structure, comments etc.

• mergeable_data

  (MutableMapping[str, object]) –

  The new values to be merged into the loaded data.

Source code in configzen/data.py

def roundtrip_update_mapping(
    self,
    roundtrip_data: Data,
    mergeable_data: MutableMapping[str, object],
) -> None:
    """
    Update the loaded data in a round-trip manner.

    Use values from the configuration altered programmatically in runtime,
    while keeping the structure and comments of the original data.

    Parameters
    ----------
    roundtrip_data
        The data to be updated. Stores the original structure, comments etc.
    mergeable_data
        The new values to be merged into the loaded data.

    """
    return roundtrip_update_mapping(
        roundtrip_data,
        mergeable_data,
        _recursive_update_mapping=self.roundtrip_update_mapping,
        _recursive_update_sequence=self.roundtrip_update_sequence,
    )

roundtrip_update_sequence(roundtrip_data, mergeable_data)

Merge new data sequence without losing comments.

Source code in configzen/data.py

def roundtrip_update_sequence(
    self,
    roundtrip_data: MutableSequence[object],
    mergeable_data: Sequence[object],
) -> None:
    """Merge new data sequence without losing comments."""
    return roundtrip_update_sequence(
        roundtrip_data,
        mergeable_data,
        _recursive_update_mapping=self.roundtrip_update_mapping,
        _recursive_update_sequence=self.roundtrip_update_sequence,
    )

__init_subclass__(*, skip_hook=False)

Subclass hook. Pass skip_hook=True to skip it.

Source code in configzen/data.py

def __init_subclass__(cls, *, skip_hook: bool = False) -> None:
    """Subclass hook. Pass skip_hook=True to skip it."""
    if not skip_hook:
        if getattr(cls, "option_name", None) is None:
            msg = (
                f"{cls.__name__} must have an option_name attribute "
                "if it is not a class with skip_hook=True parameter"
            )
            raise TypeError(msg)
        if getattr(cls, "file_extensions", None) is None:
            cls.file_extensions = set()
        cls.file_extensions.add(cls.default_extension)
        cls.register_file_extensions()

roundtrip_update_mapping(roundtrip_data, mergeable_data, *, _recursive_update_mapping=None, _recursive_update_sequence=None)

Update a mapping without losing recursively attached metadata.

Source code in configzen/data.py

def roundtrip_update_mapping(
    roundtrip_data: Data,
    mergeable_data: MutableMapping[str, object],
    *,
    _recursive_update_mapping: Callable[[Data, MutableMapping[str, object]], None]
    | None = None,
    _recursive_update_sequence: Callable[
        [MutableSequence[object], Sequence[object]],
        None,
    ]
    | None = None,
) -> None:
    """Update a mapping without losing recursively attached metadata."""
    if _recursive_update_mapping is None:
        _recursive_update_mapping = partial(
            roundtrip_update_mapping,
            _recursive_update_sequence=_recursive_update_sequence,
        )
    if _recursive_update_sequence is None:
        _recursive_update_sequence = partial(
            roundtrip_update_sequence,
            _recursive_update_mapping=_recursive_update_mapping,
        )
    for key, value in roundtrip_data.items():
        if key in mergeable_data:
            new_value = mergeable_data.pop(key)
            if isinstance(value, MutableMapping):
                # Coerce it's a dict to ensure it has the .pop() method
                _recursive_update_mapping(
                    value,
                    dict(cast("Mapping[str, object]", new_value)),
                )
            elif isinstance(value, MutableSequence):
                _recursive_update_sequence(
                    value,
                    cast("MutableSequence[object]", new_value),
                )
            else:
                roundtrip_data[key] = new_value
    for key, value in mergeable_data.items():
        roundtrip_data[key] = value

roundtrip_update_sequence(roundtrip_data, mergeable_data, *, _recursive_update_mapping=None, _recursive_update_sequence=None)

Update a sequence without losing recursively attached metadata.

Source code in configzen/data.py

def roundtrip_update_sequence(
    roundtrip_data: MutableSequence[object],
    mergeable_data: Sequence[object],
    *,
    _recursive_update_mapping: Callable[[Data, MutableMapping[str, object]], None]
    | None = None,
    _recursive_update_sequence: Callable[
        [MutableSequence[object], Sequence[object]],
        None,
    ]
    | None = None,
) -> None:
    """Update a sequence without losing recursively attached metadata."""
    if _recursive_update_mapping is None:
        _recursive_update_mapping = partial(
            roundtrip_update_mapping,
            _recursive_update_sequence=_recursive_update_sequence,
        )
    if _recursive_update_sequence is None:
        _recursive_update_sequence = partial(
            roundtrip_update_sequence,
            _recursive_update_mapping=_recursive_update_mapping,
        )
    sequence_length = len(mergeable_data)
    for i, (roundtrip_item, mergeable_item) in enumerate(
        zip_longest(
            roundtrip_data,
            mergeable_data,
        ),
    ):
        if i >= sequence_length:
            roundtrip_data[i] = roundtrip_item
        elif isinstance(roundtrip_item, MutableMapping):
            _recursive_update_mapping(
                roundtrip_item,
                dict(cast("Mapping[str, object]", mergeable_item)),
            )
        elif isinstance(roundtrip_item, MutableSequence):
            _recursive_update_sequence(
                roundtrip_item,
                cast("list[object]", mergeable_item),
            )

errors

Specialized exceptions raised by configzen.

Classes:

• ConfigError –

  Base class for all errors related to configzen.

• ConfigLoadError –

  Raised when the configuration cannot be loaded.

• ConfigReloadError –

  Raised when the configuration cannot be reloaded.

• ConfigSaveError –

  Raised when the configuration cannot be saved.

• NotAMappingError –

  Raised when the configuration being loaded is not a mapping.

• ConfigProcessorError –

  Raised when a configuration replacement processor error occurs.

• BaseRouteError –

  Raised when a configuration item route is invalid.

• RouteError –

  Raised when a configuration item route is invalid at a specific index.

• LinkedRouteError –

  Raised when a declared configuration item route is invalid.

ConfigError(message)

Bases: Exception

Base class for all errors related to configzen.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

ConfigLoadError(message)

Bases: ConfigError

Raised when the configuration cannot be loaded.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

ConfigReloadError(message)

Bases: ConfigLoadError

Raised when the configuration cannot be reloaded.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

ConfigSaveError(message)

Bases: ConfigError

Raised when the configuration cannot be saved.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

NotAMappingError(message)

Bases: ConfigLoadError, TypeError

Raised when the configuration being loaded is not a mapping.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

ConfigProcessorError(message)

Bases: ConfigError

Raised when a configuration replacement processor error occurs.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

BaseRouteError(message)

Bases: ConfigError, ValueError

Raised when a configuration item route is invalid.

Source code in configzen/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)

RouteError(message, route, index)

Bases: BaseRouteError

Raised when a configuration item route is invalid at a specific index.

Methods:

• __str__ –

  Return a string representation of the route error.

Source code in configzen/errors.py

def __init__(self, message: str, route: str, index: int) -> None:
    self.message = message
    self.route = route
    self.index = index

__str__()

Return a string representation of the route error.

Source code in configzen/errors.py

def __str__(self) -> str:
    """Return a string representation of the route error."""
    return f"{self.message} ({self.route}:{self.index})"

LinkedRouteError(message, route, config_class)

Bases: BaseRouteError

Raised when a declared configuration item route is invalid.

Methods:

• __str__ –

  Return a string representation of the route error.

Source code in configzen/errors.py

def __init__(
    self,
    message: str,
    route: str,
    config_class: type[BaseConfig],
) -> None:
    self.message = message
    self.route = route
    self.config_class = config_class

__str__()

Return a string representation of the route error.

Source code in configzen/errors.py

def __str__(self) -> str:
    """Return a string representation of the route error."""
    return f"{self.message} ({self.config_class.__name__}.{self.route})"

formats

configzen.formats: Data formats supported natively by configzen.

Modules:

• std_json –

  configzen.formats.std_json: The JSON data format.

• std_plist –

  configzen.formats.std_plist: The Plist data format.

• toml –

  configzen.formats.toml: The TOML data format.

• yaml –

  configzen.formats.yaml: The YAML data format.

std_json

configzen.formats.std_json: The JSON data format.

Classes:

• JSONOptions –

  Prototype of the allowed options for the JSON data format.

• JSONDataFormat –

  The JSON data format.

JSONOptions

Bases: DataFormatOptions

Prototype of the allowed options for the JSON data format.

JSONDataFormat

Bases: TextDataFormat[JSONOptions]

The JSON data format.

Methods:

• configure –

  For the documentation of the options, see the JSONOptions class.

• load –

  Load the JSON data from the given stream.

• dump –

  Dump the given JSON data to the given stream.

configure(**options)

For the documentation of the options, see the JSONOptions class.

Source code in configzen/formats/std_json.py

def configure(self, **options: Unpack[JSONOptions]) -> None:
    """For the documentation of the options, see the JSONOptions class."""
    self.json_encoder = JSONEncoder(
        skipkeys=options.get("skipkeys") or self.json_encoder.skipkeys,
        ensure_ascii=options.get("ensure_ascii") or self.json_encoder.ensure_ascii,
        check_circular=options.get("check_circular")
        or self.json_encoder.check_circular,
        allow_nan=options.get("allow_nan") or self.json_encoder.allow_nan,
        indent=options.get("indent") or self.json_encoder.indent,
        separators=options.get("separators")
        or (
            self.json_encoder.item_separator,
            self.json_encoder.key_separator,
        ),
        default=options.get("default") or self.json_encoder.default,
    )
    self.json_decoder = JSONDecoder(
        object_hook=options.get("object_hook") or self.json_decoder.object_hook,
        parse_float=options.get("parse_float") or self.json_decoder.parse_float,
        parse_int=options.get("parse_int") or self.json_decoder.parse_int,
        parse_constant=options.get("parse_constant")
        or self.json_decoder.parse_constant,
        strict=options.get("strict") or self.json_decoder.strict,
        object_pairs_hook=options.get("object_pairs_hook")
        or self.json_decoder.object_pairs_hook,
    )

load(stream)

Load the JSON data from the given stream.

Source code in configzen/formats/std_json.py

def load(self, stream: IO[str]) -> Data:
    """Load the JSON data from the given stream."""
    document = (
        load(
            stream,
            cls=cast("type[JSONDecoder]", lambda **_: self.json_decoder),
        )
        or {}
    )
    if not isinstance(document, dict):
        msg = (
            f"Expected a dict mapping, "
            f"but got {type(document).__name__} instead."
        )
        raise TypeError(msg)
    return document

dump(data, stream)

Dump the given JSON data to the given stream.

Source code in configzen/formats/std_json.py

def dump(self, data: Data, stream: IO[str]) -> None:
    """Dump the given JSON data to the given stream."""
    dump(
        data,
        stream,
        cls=cast("type[JSONEncoder]", lambda **_: self.json_encoder),
    )

std_plist

configzen.formats.std_plist: The Plist data format.

Classes:

• PlistOptions –

  Prototype of the allowed options for the Plist data format.

• PlistDataFormat –

  The Plist data format.

PlistOptions

Bases: DataFormatOptions

Prototype of the allowed options for the Plist data format.

PlistDataFormat

Bases: BinaryDataFormat[PlistOptions]

The Plist data format.

Methods:

• configure –

  For the documentation of the options, see the PlistOptions class.

• load –

  Load the data from the given stream.

• dump –

  Dump the given data to the stream.

configure(**options)

For the documentation of the options, see the PlistOptions class.

Source code in configzen/formats/std_plist.py

def configure(self, **options: Unpack[PlistOptions]) -> None:
    """For the documentation of the options, see the PlistOptions class."""
    if "fmt" not in options:
        options["fmt"] = self.plist_options.get("fmt", PlistFormat.FMT_XML)
    if "dict_type" not in options:
        options["dict_type"] = self.plist_options.get(
            "dict_type",
            dict,
        )  # type: ignore[typeddict-item]
    if "sort_keys" not in options:
        # configzen focuses on preserving the original structure,
        # so we don't sort by default.
        options["sort_keys"] = self.plist_options.get("sort_keys", False)
    if "skipkeys" not in options:
        options["skipkeys"] = self.plist_options.get("skipkeys", False)
    self.plist_options = options

load(stream)

Load the data from the given stream.

Source code in configzen/formats/std_plist.py

def load(self, stream: IO[bytes]) -> Data:
    """Load the data from the given stream."""
    dict_class: type[MutableMapping[str, Any]] = self.plist_options["dict_type"]
    document = (
        load(
            stream,
            fmt=self.plist_options["fmt"],
            dict_type=dict_class,
        )
        or dict_class()
    )
    if not isinstance(document, dict_class):
        msg = (
            f"Expected a {dict_class.__name__} mapping, "
            f"but got {type(document).__name__} instead."
        )
        raise TypeError(msg)
    return document

dump(data, stream)

Dump the given data to the stream.

Source code in configzen/formats/std_plist.py

def dump(self, data: Data, stream: IO[bytes]) -> None:
    """Dump the given data to the stream."""
    dump(
        data,
        stream,
        fmt=self.plist_options["fmt"],
        sort_keys=self.plist_options["sort_keys"],
        skipkeys=self.plist_options["skipkeys"],
    )

toml

configzen.formats.toml: The TOML data format.

Classes:

• TOMLOptions –

  Prototype of the allowed options for the TOML data format.

• TOMLDataFormat –

  The TOML data format.

TOMLOptions

Bases: DataFormatOptions

Prototype of the allowed options for the TOML data format.

Attributes:

• encoders (list[Encoder]) –

  List of encoders to perform automatic tomlkit.register_encoder() calls on.

• unregister_old_encoders (bool) –

  Whether to unregister all previously registered encoders

• sort_keys (bool) –

  Whether to sort keys in the output.

encoders: list[Encoder]

List of encoders to perform automatic tomlkit.register_encoder() calls on.

unregister_old_encoders: bool

Whether to unregister all previously registered encoders before registering the new ones.

sort_keys: bool

Whether to sort keys in the output.

TOMLDataFormat

Bases: TextDataFormat[TOMLOptions]

The TOML data format.

Methods:

• configure –

  For the documentation of the options, see the TOMLOptions class.

• load –

  Load the data from the given stream.

• dump –

  Dump the data to the given stream.

configure(**options)

For the documentation of the options, see the TOMLOptions class.

Source code in configzen/formats/toml.py

def configure(self, **options: Unpack[TOMLOptions]) -> None:
    """For the documentation of the options, see the TOMLOptions class."""
    old_options = self.toml_options
    toml_encoders = options.get("encoders") or old_options.get("encoders") or []
    cleanup_old_encoders = options.get("cleanup_old_encoders", False)

    if cleanup_old_encoders:
        for encoder in self.toml_options.get("encoders") or []:
            unregister_encoder(encoder)

    for encoder in toml_encoders:
        register_encoder(encoder)

load(stream)

Load the data from the given stream.

Source code in configzen/formats/toml.py

def load(self, stream: IO[str]) -> Data:
    """Load the data from the given stream."""
    return load(stream)

dump(data, stream)

Dump the data to the given stream.

Source code in configzen/formats/toml.py

def dump(self, data: Data, stream: IO[str]) -> None:
    """Dump the data to the given stream."""
    dump(
        data,
        stream,
        sort_keys=self.toml_options.get("sort_keys", False),
    )

yaml

configzen.formats.yaml: The YAML data format.

Classes:

• YAMLOptions –

  Prototype of the allowed options for the YAML data format.

• YAMLDataFormat –

  The YAML data format.

YAMLOptions

Bases: DataFormatOptions

Prototype of the allowed options for the YAML data format.

For more information, see the documentation of the ruamel.yaml.YAML class.

Attributes:

• classes (list[type]) –

  List of classes to automatically call YAML.register_class() on.

• version (str | tuple[int | str, ...]) –

  The YAML version to use.

• indent (int) –

  Indentation width.

• block_seq_indent (int) –

  Indentation for nested block sequences.

classes: list[type]

List of classes to automatically call YAML.register_class() on.

version: str | tuple[int | str, ...]

The YAML version to use.

indent: int

Indentation width.

block_seq_indent: int

Indentation for nested block sequences.

YAMLDataFormat

Bases: TextDataFormat[YAMLOptions]

The YAML data format.

Methods:

• configure –

  For the documentation of the options, see the YAMLOptions class.

• load –

  Load the data from a stream.

• dump –

  Load the data from a stream.

configure(**options)

For the documentation of the options, see the YAMLOptions class.

Source code in configzen/formats/yaml.py

def configure(self, **options: Unpack[YAMLOptions]) -> None:
    """For the documentation of the options, see the YAMLOptions class."""
    yaml_classes = options.pop("classes", None) or []

    old_yaml = self.yaml
    yaml_version = options.pop("version", None) or old_yaml.version
    yaml_indent = options.pop("indent", None) or old_yaml.old_indent
    yaml_block_seq_indent = (
        options.pop("block_seq_indent", None) or old_yaml.block_seq_indent
    )

    yaml = YAML(**options)  # type: ignore[arg-type,misc]
    yaml.version = yaml_version  # type: ignore[assignment]
    yaml.indent = yaml_indent
    yaml.block_seq_indent = yaml_block_seq_indent

    for cls in yaml_classes:
        yaml.register_class(cls)

    self.yaml = yaml

load(stream)

Load the data from a stream.

Return a mutable mapping representing the loaded data which is mutation-sensitive (for round-trip processing).

Every configuration source transforms the input data into a stream to be processed by the data format, because most data format libraries operate on streams.

This method is called by the configuration model.

Source code in configzen/formats/yaml.py

def load(self, stream: IO[str]) -> Data:
    """
    Load the data from a stream.

    Return a mutable mapping representing the loaded data
    which is mutation-sensitive (for round-trip processing).

    Every configuration source transforms the input data into a stream
    to be processed by the data format, because most data format libraries
    operate on streams.

    This method is called by the configuration model.
    """
    data = self.yaml.load(stream) or CommentedMap()
    if not isinstance(data, dict):
        msg = f"Expected a dict, but got {type(data).__name__}."
        raise TypeError(msg)
    return data

dump(data, stream)

Load the data from a stream.

Every configuration source transforms the input data into a stream to be processed by the data format, because most data format libraries operate on streams.

This method is called by the configuration model.

Source code in configzen/formats/yaml.py

def dump(self, data: Data, stream: IO[str]) -> None:
    """
    Load the data from a stream.

    Every configuration source transforms the input data into a stream
    to be processed by the data format, because most data format libraries
    operate on streams.

    This method is called by the configuration model.
    """
    self.yaml.dump(data, stream)

module_proxy

Runtime modules with attribute type validation.

Classes:

• ModuleProxy –

  Proxy object that extends a runtime module with type validation.

ModuleProxy(name, config, module_namespace=None, doc=None)

Bases: ModuleType, Generic[ConfigObject]

Proxy object that extends a runtime module with type validation.

Triggered via a config instance (initialization and assignment).

Parameters:

• name

  (str) –

  The name of the module.

• config

  (ConfigObject) –

  The configuration model to use for type validation.

• module_namespace

  (dict[str, Any] | None, default: None ) –

  The module namespace to wrap.

Methods:

• __getattribute__ –

  Get an attribute of the underlying model.

• __setattr__ –

  Set an attribute on the underlying model.

• __repr__ –

  Get the string representation of the module proxy.

• get_config –

  Get the configuration model.

• wrap_module –

  Wrap a module to ensure type validation.

• wrap_this_module –

  Wrap the module calling this function.

Source code in configzen/module_proxy.py

def __init__(
    self,
    name: str,
    config: ConfigObject,
    module_namespace: dict[str, Any] | None = None,
    doc: str | None = None,
) -> None:
    object.__setattr__(self, "__config__", config)
    object.__setattr__(self, "__locals__", module_namespace or {})
    object.__setattr__(config, "__wrapped_module__", self)

    super().__init__(name=name, doc=doc)

    parts = name.split(".")
    if len(parts) > 1:
        # Set the proxy module as an attribute of its parent.
        parent = sys.modules[".".join(parts[:-1])]
        setattr(parent, parts[-1], self)

    # Make reusable.
    sys.modules[name] = self

__getattribute__(name)

Get an attribute of the underlying model.

Source code in configzen/module_proxy.py

def __getattribute__(self, name: str) -> Any:
    """Get an attribute of the underlying model."""
    if _is_dunder(name):
        return object.__getattribute__(self, name)

    config = self.__config__
    try:
        return getattr(config, name)
    except AttributeError:
        try:
            return self.__locals__[name]
        except KeyError:
            return object.__getattribute__(self, name)

__setattr__(key, value)

Set an attribute on the underlying model.

Source code in configzen/module_proxy.py

def __setattr__(self, key: str, value: Any) -> None:
    """Set an attribute on the underlying model."""
    config = self.get_config()
    if not _is_dunder(key) and key in config.model_fields:
        setattr(config, key, value)
    self.__locals__[key] = value

__repr__()

Get the string representation of the module proxy.

Inform the user that this is a configuration module.

Source code in configzen/module_proxy.py

def __repr__(self) -> str:
    """
    Get the string representation of the module proxy.

    Inform the user that this is a configuration module.
    """
    return super().__repr__().replace("module", "configuration module", 1)

get_config()

Get the configuration model.

Source code in configzen/module_proxy.py

def get_config(self) -> ConfigObject:
    """Get the configuration model."""
    return self.__config__

wrap_module(module_name, config_class=None, namespace=None, /, **values)

Wrap a module to ensure type validation.

Every attribute of the wrapped module that is also a field of the config will be validated against it. The module will be extended with the config's attributes. Assignments on the module's attributes will be propagated to the configuration instance. It could be said that the module becomes a proxy for the configuration once wrapped.

Parameters:

• module_name

  (str) –

  The name of the module to wrap.

• config_class

  (type[ConfigObject] | None, default: None ) –

  The config class to use for type validation.

• namespace

  (dict[str, Any] | None, default: None ) –

  The namespace of the module to wrap. If not provided, it will be retrieved from sys.modules.

• values

  (Any, default: {} ) –

  Values used to initialize the config.

Returns:

• The wrapped module. –

Source code in configzen/module_proxy.py

@classmethod
def wrap_module(
    cls,
    module_name: str,
    config_class: type[ConfigObject] | None = None,
    namespace: dict[str, Any] | None = None,
    /,
    **values: Any,
) -> ModuleProxy[ConfigObject]:
    """
    Wrap a module to ensure type validation.

    Every attribute of the wrapped module that is also a field of the config will be
    validated against it. The module will be extended with the config's attributes.
    Assignments on the module's attributes will be propagated to the configuration
    instance. It could be said that the module becomes a proxy for the configuration
    once wrapped.

    Parameters
    ----------
    module_name
        The name of the module to wrap.
    config_class
        The config class to use for type validation.
    namespace
        The namespace of the module to wrap. If not provided, it will be
        retrieved from `sys.modules`.
    values
        Values used to initialize the config.

    Returns
    -------
    The wrapped module.

    """
    from configzen.config import BaseConfig

    if namespace is None:
        module_namespace = vars(sys.modules[module_name])
    else:
        module_namespace = namespace

    if config_class is None:

        class ConfigModule(BaseConfig):
            __module__ = module_name
            __annotations__ = module_namespace["__annotations__"]
            for key in __annotations__:
                locals()[key] = module_namespace[key]

        config_class = cast("type[ConfigObject]", ConfigModule)

    module_values = {}
    field_names = frozenset(
        field_info.validation_alias
        or field_info.alias
        or field_info.title
        or field_name
        for field_name, field_info in config_class.model_fields.items()
    )
    for key, value in module_namespace.items():
        if key in field_names:
            module_values[key] = value
    config = config_class.model_validate({**module_values, **values})

    return cls(
        config=config,
        module_namespace=module_namespace,
        name=module_namespace.get("__name__") or module_name,
        doc=module_namespace.get("__doc__"),
    )

wrap_this_module(config_class=None, /, **values)

Wrap the module calling this function.

For more information on wrapping modules, see ModuleProxy.wrap_module().

Parameters:

• config_class

  (type[ConfigObject] | None, default: None ) –

  The config class to use for type validation.

• values

  (Any, default: {} ) –

  Values used to initialize the config.

Source code in configzen/module_proxy.py

@classmethod
def wrap_this_module(
    cls,
    config_class: type[ConfigObject] | None = None,
    /,
    **values: Any,
) -> ModuleProxy[ConfigObject]:
    """
    Wrap the module calling this function.

    For more information on wrapping modules, see `ModuleProxy.wrap_module()`.

    Parameters
    ----------
    config_class
        The config class to use for type validation.
    values
        Values used to initialize the config.

    """
    current_frame = inspect.currentframe()
    if current_frame is None:
        msg = "Could not get the current frame"
        raise RuntimeError(msg)
    frame_back = current_frame.f_back
    if frame_back is None:
        msg = "Could not get the frame back"
        raise RuntimeError(msg)
    return cls.wrap_module(
        {**frame_back.f_globals, **frame_back.f_locals}["__name__"],
        config_class,
        {**frame_back.f_locals, **values},
    )

processor

Replacement API processor for configuration data.

Allows to tweak the configuration data programmatically before it is given to the model config and revert the changes back to the original data structure when the configuration managed by that model is saved.

Classes:

• ProcessorOptions –

  Prototype of the allowed options for the ConfigProcessor class.

• ConfigProcessor –

  A class that takes in configuration data and processes it.

• ProcessorReplacement –

  A change that was made to the configuration data during processing.

• FileSystemAwareConfigProcessor –

  Config processor that is aware of the file system.

ProcessorOptions

Bases: TypedDict

Prototype of the allowed options for the ConfigProcessor class.

ConfigProcessor(initial, *, macro_prefix=Char('^'), update_prefix=Char('+'), macros_on_top=False, lenient=True)

A class that takes in configuration data and processes it.

Recursively resolves & applies replacements in data magically.

Methods:

• create_processor –

  Create a new configuration processor with identical options.

• get_processed_data –

  Create the data with replacements or return the one already cached.

• __init_subclass__ –

  Merge macro registries on subclass.

• sanitize_macro_name –

  Ensure a uniform name of every macro.

• macro –

  Override a macro.

Attributes:

• macros (MacroDict) –

  Get macros bound to this processor.

• roundtrip_initial (Data) –

  The initial configuration data that the processor was given.

Source code in configzen/processor.py

def __init__(  # noqa: PLR0913
    self,
    initial: Data,
    *,
    macro_prefix: Char = Char("^"),  # noqa: B008
    update_prefix: Char = Char("+"),  # noqa: B008
    macros_on_top: bool = False,
    lenient: bool = True,
) -> None:
    self.__initial = initial
    self.__data: _ProcessedData = None  # type: ignore[assignment]

    self.options = ProcessorOptions(
        macro_prefix=macro_prefix,
        update_prefix=update_prefix,
        macros_on_top=macros_on_top,
        lenient=lenient,
    )

macros: MacroDict

Get macros bound to this processor.

roundtrip_initial: Data

The initial configuration data that the processor was given.

create_processor(data)

Create a new configuration processor with identical options.

Source code in configzen/processor.py

def create_processor(self, data: Data) -> ConfigProcessor:
    """Create a new configuration processor with identical options."""
    return type(self)(data, **self.options)

get_processed_data(*, force=False)

Create the data with replacements or return the one already cached.

Parameters:

• force

  (bool, default: False ) –

  Whether to forcibly parse the original data even if it was already parsed. Default is False.

Source code in configzen/processor.py

def get_processed_data(
    self,
    *,
    force: bool = False,
) -> _ProcessedData:
    """
    Create the data with replacements or return the one already cached.

    Parameters
    ----------
    force
        Whether to forcibly parse the original data even if it was already parsed.
        Default is False.

    """
    if force or self.__data is None:
        self.__data = self._get_processed_data(
            data=self.__initial,
            options=self.options,
            macros=self.macros,
        )
    return self.__data

__init_subclass__()

Merge macro registries on subclass.

Source code in configzen/processor.py

def __init_subclass__(cls) -> None:
    """Merge macro registries on subclass."""
    macros_from_class_dict = {
        macro_name: func
        for func in vars(cls).values()
        if (macro_name := getattr(func, MACRO_FUNC, None))
    }
    try:
        macros = {**getattr(cls.__base__, "_macros", {}), **macros_from_class_dict}
    except AttributeError:
        macros = {}
    cls._macros = macros

sanitize_macro_name(name)

Ensure a uniform name of every macro.

Source code in configzen/processor.py

@staticmethod
def sanitize_macro_name(name: str) -> str:
    """Ensure a uniform name of every macro."""
    return name.strip().casefold()

macro(name, macro)

Override a macro.

Source code in configzen/processor.py

@classmethod
def macro(cls, name: str, macro: MacroT) -> MacroT:
    """Override a macro."""
    name = cls.sanitize_macro_name(name)
    cls._macros[name] = macro
    return macro

ProcessorReplacement(key, value, content)

A change that was made to the configuration data during processing.

Attributes:

• key (str) –

  The key of the item before alteration.

• value (object) –

  The value of the item before alteration.

• content (Data) –

  The value to unpack in place of the alteration key.

FileSystemAwareConfigProcessor(initial, *, macro_prefix=Char('^'), update_prefix=Char('+'), macros_on_top=False, lenient=True)

Bases: ConfigProcessor

Config processor that is aware of the file system.

Can handle requests for transcluding other configuration files to achieve a sense of extendability.

Methods:

• extend –

  Transclude a config in this config.

Source code in configzen/processor.py

def __init__(  # noqa: PLR0913
    self,
    initial: Data,
    *,
    macro_prefix: Char = Char("^"),  # noqa: B008
    update_prefix: Char = Char("+"),  # noqa: B008
    macros_on_top: bool = False,
    lenient: bool = True,
) -> None:
    self.__initial = initial
    self.__data: _ProcessedData = None  # type: ignore[assignment]

    self.options = ProcessorOptions(
        macro_prefix=macro_prefix,
        update_prefix=update_prefix,
        macros_on_top=macros_on_top,
        lenient=lenient,
    )

extend(sources)

Transclude a config in this config.

Source code in configzen/processor.py

@macro
def extend(self, sources: str | dict[str, str]) -> Data:
    """Transclude a config in this config."""
    if isinstance(sources, str):
        source = get_config_source(sources)
        return source.load()
    return {
        key: get_config_source(source).load() for key, source in sources.items()
    }

routes

Routes creation and parsing.

Classes:

• Step –

  A configuration route step.

• GetAttr –

  A route step that gets an attribute from an object.

• GetItem –

  A route step that gets an item from an object.

• Route –

  Routes are, lists of steps that are used to access values in a configuration.

Functions:

• advance_linked_route –

  Move one step forward in a linked route.

Step(key)

Bases: Generic[_KT]

A configuration route step.

Do not use this class directly. Use GetAttr or GetItem instead.

Methods:

• __eq__ –

  Compare this step to another step.

• get –

  Perform a get operation.

• set –

  Perform a set operation.

• __call__ –

  Perform a get operation.

• __repr__ –

  Represent this step in a string.

Source code in configzen/routes.py

def __init__(self, key: _KT, /) -> None:
    self.key = key

__eq__(other)

Compare this step to another step.

Source code in configzen/routes.py

def __eq__(self, other: object) -> bool:
    """Compare this step to another step."""
    if isinstance(other, Step):
        return (
            issubclass(type(other), type(self))
            or issubclass(type(self), type(other))
        ) or self.key == other.key
    return NotImplemented

get(_)

Perform a get operation.

Source code in configzen/routes.py

def get(self, _: Any, /) -> object:
    """Perform a get operation."""
    raise NotImplementedError

set(_, __)

Perform a set operation.

Source code in configzen/routes.py

def set(self, _: Any, __: object, /) -> None:
    """Perform a set operation."""
    raise NotImplementedError

__call__(obj)

Perform a get operation.

Source code in configzen/routes.py

def __call__(self, obj: Any, /) -> object:
    """Perform a get operation."""
    return self.get(obj)

__repr__()

Represent this step in a string.

Source code in configzen/routes.py

def __repr__(self) -> str:
    """Represent this step in a string."""
    return f"{type(self).__name__}({self.key!r})"

GetAttr(key)

Bases: Step[str]

A route step that gets an attribute from an object.

The argument is used as an attribute name.

Methods:

• get –

  Get an attribute from an object.

• set –

  Set an attribute in an object.

• __str__ –

  Compose this step into a string.

Source code in configzen/routes.py

def __init__(self, key: _KT, /) -> None:
    self.key = key

get(target)

Get an attribute from an object.

Source code in configzen/routes.py

def get(self, target: Any, /) -> object:
    """Get an attribute from an object."""
    return getattr(target, self.key)

set(target, value)

Set an attribute in an object.

Source code in configzen/routes.py

def set(self, target: Any, value: object, /) -> None:
    """Set an attribute in an object."""
    setattr(target, self.key, value)

__str__()

Compose this step into a string.

Source code in configzen/routes.py

def __str__(self) -> str:
    """Compose this step into a string."""
    return str(self.key).replace(Route.TOKEN_DOT, r"\.")

GetItem(key, /, *, ignore_digit=False)

Bases: Step[Union[int, str]]

A route step that gets an item from an object.

If the argument is a string, it is used checked for being a digit. Unless explicitly escaped, if it is a digit, it is casted to an integer. Otherwise, it is used as is.

Methods:

• get –

  Get an item from an object.

• set –

  Set an item in an object.

• __str__ –

  Compose this step into a string.

Source code in configzen/routes.py

def __init__(self, key: int | str, /, *, ignore_digit: bool = False) -> None:
    self.escape = False
    if isinstance(key, str) and key.isdigit():
        if ignore_digit:
            self.escape = True
        else:
            key = int(key)
    super().__init__(key)

get(target)

Get an item from an object.

Source code in configzen/routes.py

def get(self, target: Any, /) -> object:
    """Get an item from an object."""
    return target[self.key]

set(target, value)

Set an item in an object.

Source code in configzen/routes.py

def set(self, target: Any, value: object, /) -> None:
    """Set an item in an object."""
    target[self.key] = value

__str__()

Compose this step into a string.

Source code in configzen/routes.py

def __str__(self) -> str:
    """Compose this step into a string."""
    argument = str(self.key)
    if self.escape:
        argument = Route.TOKEN_ESCAPE + argument
    return argument.join(
        Route.TOKEN_ENTER + Route.TOKEN_LEAVE,
    ).replace(Route.TOKEN_DOT, r"\.")

Route(route, *, allow_empty=False)

Routes are, lists of steps that are used to access values in a configuration.

Each step is either a key or an index.

A route can be created from a string, a list of steps, or another route.

Examples:

>>> route = Route("a.b.c")
>>> route
<Route 'a.b.c'>
>>> route.steps
[GetAttr('a'), GetAttr('b'), GetAttr('c')]

Parameters:

• route

  (RouteLike) –

  A route to parse.

• allow_empty

  (bool, default: False ) –

  Whether to allow empty routes.

Methods:

• __hash__ –

  Get a hash of this route.

• parse –

  Parse a route into steps.

• decompose –

  Decompose a route into a list of steps.

• compose –

  Compose this route into a string.

• enter –

  Enter a subroute.

• get –

  Get an object at the end of this route.

• set –

  Set an object pointed to by this route.

• __eq__ –

  Compare this route to another route.

• __str__ –

  Compose this route into a string.

• __iter__ –

  Yield all steps in this route.

• __repr__ –

  Represent this route in a string.

Attributes:

• steps (list[Step[Any]]) –

  Get all steps in this route.

Source code in configzen/routes.py

def __init__(
    self,
    route: RouteLike,
    *,
    allow_empty: bool = False,
) -> None:
    steps = self.parse(route)
    if not (allow_empty or steps):
        msg = "Empty configuration route"
        raise ValueError(msg)
    self.__steps = tuple(steps)

steps: list[Step[Any]]

Get all steps in this route.

__hash__()

Get a hash of this route.

Source code in configzen/routes.py

def __hash__(self) -> int:
    """Get a hash of this route."""
    return hash(self.__steps)

parse(route)

Parse a route into steps.

Parameters:

• route

  (RouteLike) –

  The route to parse.

Returns:

• List of steps. –

Source code in configzen/routes.py

@classmethod
def parse(cls, route: RouteLike) -> list[Step[Any]]:
    """
    Parse a route into steps.

    Parameters
    ----------
    route
        The route to parse.

    Returns
    -------
    List of steps.

    """
    if isinstance(route, Step):
        return [route]
    if isinstance(route, Route):
        return route.steps
    if isinstance(route, (tuple, list)):
        patched_route: list[Step[Any]] = []
        for element in route:
            if isinstance(element, (str, int)):
                try:
                    patched_element = next(iter(cls.parse(element)))
                except StopIteration:
                    continue
            else:
                patched_element = element
            patched_route.append(patched_element)
        return patched_route
    if isinstance(route, int):
        return [GetItem(route)]
    if isinstance(route, str):
        return cls.decompose(route)
    msg = f"Invalid route type {type(route)!r}"
    raise TypeError(msg)

decompose(route)

Decompose a route into a list of steps.

Parameters:

• route

  (str) –

  A route to decompose.

Returns:

• List of steps. –

Source code in configzen/routes.py

@classmethod
def decompose(cls, route: str) -> list[Step[Any]]:
    """
    Decompose a route into a list of steps.

    Parameters
    ----------
    route
        A route to decompose.

    Returns
    -------
    List of steps.

    """
    if not route:
        return []

    dot, escape, enter, leave = cls.TOKENS

    return _route_decompose(
        route,
        dot=dot,
        escape=escape,
        enter=enter,
        leave=leave,
    )

compose()

Compose this route into a string.

Source code in configzen/routes.py

def compose(self) -> str:
    """Compose this route into a string."""
    composed = ""
    steps = self.__steps
    for index, step in enumerate(steps):
        composed += str(step)
        if index < len(steps) - 1:
            ahead = steps[index + 1]
            if isinstance(ahead, GetAttr):
                composed += self.TOKEN_DOT
    return composed

enter(subroute)

Enter a subroute.

Parameters:

• subroute

  (RouteLike) –

  A subroute to enter.

Source code in configzen/routes.py

def enter(self, subroute: RouteLike) -> Route:
    """
    Enter a subroute.

    Parameters
    ----------
    subroute
        A subroute to enter.

    """
    return type(self)(self.steps + self.parse(subroute))

get(obj)

Get an object at the end of this route.

Parameters:

• obj

  (Any) –

  An object to dive in.

Returns:

• The result of visiting the object. –

Source code in configzen/routes.py

def get(self, obj: Any, /) -> object:
    """
    Get an object at the end of this route.

    Parameters
    ----------
    obj
        An object to dive in.

    Returns
    -------
    The result of visiting the object.

    """
    return reduce(lambda obj, step: step(obj), self.__steps, obj)

set(obj, value)

Set an object pointed to by this route.

Parameters:

• obj

  (Any) –

  An object to dive in.

• value

  (object) –

  A value to set.

Returns:

• The result of visiting the object. –

Source code in configzen/routes.py

def set(self, obj: Any, value: object, /) -> None:
    """
    Set an object pointed to by this route.

    Parameters
    ----------
    obj
        An object to dive in.

    value
        A value to set.

    Returns
    -------
    The result of visiting the object.

    """
    steps = self.steps
    last_step = steps.pop()
    last_step.set(
        reduce(lambda obj, step: step(obj), steps, obj),
        value,
    )

__eq__(other)

Compare this route to another route.

Parameters:

• other

  (object) –

  Another route to compare to.

Source code in configzen/routes.py

def __eq__(self, other: object) -> bool:
    """
    Compare this route to another route.

    Parameters
    ----------
    other
        Another route to compare to.

    """
    if isinstance(other, Route):
        return self.steps == other.steps
    if isinstance(other, str):
        return self.steps == self.decompose(other)
    if isinstance(other, (tuple, list)):
        return self.steps == self.parse(other)
    return NotImplemented

__str__()

Compose this route into a string.

Source code in configzen/routes.py

def __str__(self) -> str:
    """Compose this route into a string."""
    return self.compose()

__iter__()

Yield all steps in this route.

Source code in configzen/routes.py

def __iter__(self) -> Iterator[Step[Any]]:
    """Yield all steps in this route."""
    yield from self.__steps

__repr__()

Represent this route in a string.

Source code in configzen/routes.py

def __repr__(self) -> str:
    """Represent this route in a string."""
    return f"<{type(self).__name__} {self.compose()!r}>"

advance_linked_route(_current_head, _annotation, _step)

Move one step forward in a linked route.

Source code in configzen/routes.py

@class_singledispatch
def advance_linked_route(
    _current_head: Type[object],  # noqa: UP006
    _annotation: Any,
    _step: Step[object],
) -> Any:
    """Move one step forward in a linked route."""
    return _AnyHead

sources

Sources and destinations that hold the configuration data.

Classes:

• ConfigSource –

  Core interface for loading and saving configuration data.

• StreamConfigSource –

  A configuration source that is a stream.

• FileConfigSource –

  A configuration source that is a file.

Functions:

• get_config_source –

  Get a dedicated interface for a configuration source.

• get_stream_config_source –

  Get a dedicated interface for a configuration source stream.

• get_file_config_source –

  Get a dedicated interface for a configuration source file.

ConfigSource(source, data_format=None, **options)

Bases: Generic[SourceType, AnyStr]

Core interface for loading and saving configuration data.

If you need to implement your own configuration source class, implement a subclass of this class and pass in to the .config_load() method of your configuration or its model_config.

Methods:

• is_binary –

  Determine whether the configuration source is binary.

• load –

  Load the configuration source.

• load_async –

  Load the configuration source asynchronously.

• dump –

  Dump the configuration source.

• dump_async –

  Dump the configuration source asynchronously.

Attributes:

• data_format (DataFormat[Any, AnyStr]) –

  The current data format for a configuration source.

Source code in configzen/sources.py

def __init__(
    self,
    source: SourceType,
    data_format: str | DataFormat[Any, AnyStr] | None = None,
    **options: Unpack[FormatOptions],
) -> None:
    self._temp_stream_factory: Callable[..., IO[AnyStr]] = (
        self._binary_stream_factory
        if self.is_binary()
        else self._string_stream_factory
    )
    self.source = source
    self.options = options
    self.data_format = data_format  # type: ignore[assignment]

data_format: DataFormat[Any, AnyStr]

The current data format for a configuration source.

is_binary()

is_binary() -> Literal[False]

is_binary() -> Literal[True]

Determine whether the configuration source is binary.

Source code in configzen/sources.py

def is_binary(self: ConfigSource[SourceType, AnyStr]) -> bool:
    """Determine whether the configuration source is binary."""
    return not type_check(self, ConfigSource[Any, str])

load()

Load the configuration source.

Return its contents as a dictionary.

Source code in configzen/sources.py

@abstractmethod
def load(self) -> Data:
    """
    Load the configuration source.

    Return its contents as a dictionary.
    """
    raise NotImplementedError

load_async()

Load the configuration source asynchronously.

Return its contents as a dictionary.

Source code in configzen/sources.py

@abstractmethod
async def load_async(self) -> Data:
    """
    Load the configuration source asynchronously.

    Return its contents as a dictionary.
    """
    raise NotImplementedError

dump(data)

Dump the configuration source.

Source code in configzen/sources.py

@abstractmethod
def dump(self, data: Data) -> None:
    """Dump the configuration source."""
    raise NotImplementedError

dump_async(data)

Dump the configuration source asynchronously.

Source code in configzen/sources.py

@abstractmethod
async def dump_async(self, data: Data) -> int:
    """Dump the configuration source asynchronously."""
    raise NotImplementedError

StreamConfigSource(source, data_format, **options)

Bases: Generic[AnyStr], ConfigSource[IO[Any], Any]

A configuration source that is a stream.

Parameters:

• source

  (IO[AnyStr]) –

  The stream to the configuration source.

Methods:

• load –

  Load the configuration source.

• load_async –

  Unsupported.

• dump –

  Dump the configuration source.

• dump_async –

  Unsupported.

Source code in configzen/sources.py

def __init__(
    self,
    source: IO[AnyStr],
    data_format: str | DataFormat[Any, AnyStr],
    **options: Unpack[FormatOptions],
) -> None:
    super().__init__(source, data_format=data_format, **options)

load()

Load the configuration source.

Return its contents as a dictionary.

Source code in configzen/sources.py

def load(self) -> Data:
    """
    Load the configuration source.

    Return its contents as a dictionary.
    """
    return self.data_format.load(self.source)

load_async()

Unsupported.

Source code in configzen/sources.py

def load_async(self) -> Never:
    """Unsupported."""
    msg = "async streams are not supported for `StreamConfigSource`"
    raise NotImplementedError(msg)

dump(data)

Dump the configuration source.

Source code in configzen/sources.py

def dump(self, data: Data) -> None:
    """Dump the configuration source."""
    self.data_format.dump(data, self.source)

dump_async(_data)

Unsupported.

Source code in configzen/sources.py

def dump_async(self, _data: Data) -> Never:
    """Unsupported."""
    msg = "async streams are not supported for `StreamConfigSource`"
    raise NotImplementedError(msg)

FileConfigSource(source, data_format=None, *, use_processing_trace=True, **options)

Bases: Generic[AnyStr], ConfigSource[Path, AnyStr]

A configuration source that is a file.

Parameters:

• source

  (str | bytes | PathLike[str] | PathLike[bytes]) –

  The path to the configuration source file.

Methods:

• load –

  Load the configuration source file.

• load_async –

  Load the configuration source file asynchronously.

• dump –

  Dump the configuration data to the source file.

• dump_async –

  Load the configuration source file asynchronously.

• read –

  Read the configuration source and return its contents.

• read_async –

  Read the configuration source file asynchronously and return its contents.

• write –

  Write the configuration source file and return the number of bytes written.

• write_async –

  Write the configuration source file asynchronously.

Attributes:

• paths (list[Path]) –

  List possible path variants basing on the processing context trace.

Source code in configzen/sources.py

def __init__(
    self,
    source: str | bytes | PathLike[str] | PathLike[bytes],
    data_format: str | DataFormat[Any, Any] | None = None,
    *,
    use_processing_trace: bool = True,
    **options: Unpack[FormatOptions],
) -> None:
    super().__init__(_make_path(source), data_format=data_format, **options)
    self._use_processing_trace = use_processing_trace

paths: list[Path]

List possible path variants basing on the processing context trace.

load()

Load the configuration source file.

Return its contents as a dictionary.

Source code in configzen/sources.py

def load(self) -> Data:
    """
    Load the configuration source file.

    Return its contents as a dictionary.
    """
    data = self.data_format.load(self._temp_stream_factory(self.read()))
    self._after_load()
    return data

load_async()

Load the configuration source file asynchronously.

Return its contents as a dictionary.

Source code in configzen/sources.py

async def load_async(self) -> Data:
    """
    Load the configuration source file asynchronously.

    Return its contents as a dictionary.
    """
    data = self.data_format.load(self._temp_stream_factory(await self.read_async()))
    self._after_load()
    return data

dump(data)

Dump the configuration data to the source file.

Parameters:

• data

  (Data) –

  The data to dump to the configuration source.

Source code in configzen/sources.py

def dump(self, data: Data) -> None:
    """
    Dump the configuration data to the source file.

    Parameters
    ----------
    data
        The data to dump to the configuration source.

    """
    temp_stream = self._temp_stream_factory()
    self.data_format.dump(data, temp_stream)
    temp_stream.seek(0)
    self.write(temp_stream.read())

dump_async(data)

Load the configuration source file asynchronously.

Return its contents as a dictionary.

Parameters:

• data

  (Data) –

  The data to dump to the configuration source.

Source code in configzen/sources.py

async def dump_async(self, data: Data) -> int:
    """
    Load the configuration source file asynchronously.

    Return its contents as a dictionary.

    Parameters
    ----------
    data
        The data to dump to the configuration source.

    """
    temp_stream = self._temp_stream_factory()
    self.data_format.dump(data, temp_stream)
    temp_stream.seek(0)
    return await self.write_async(temp_stream.read())

read()

Read the configuration source and return its contents.

Source code in configzen/sources.py

def read(self) -> AnyStr:
    """Read the configuration source and return its contents."""
    errors = []
    reader = Path.read_bytes if self.is_binary() else Path.read_text
    for path in self.paths:
        try:
            return reader(path)
        except FileNotFoundError as e:  # noqa: PERF203
            errors.append(e)
            continue
    raise FileNotFoundError(errors)

read_async()

Read the configuration source file asynchronously and return its contents.

Source code in configzen/sources.py

async def read_async(self) -> AnyStr:
    """Read the configuration source file asynchronously and return its contents."""
    errors = []
    reader = AsyncPath.read_bytes if self.is_binary() else AsyncPath.read_text
    for path in map(AsyncPath, self.paths):
        try:
            return await reader(path)
        except FileNotFoundError as e:  # noqa: PERF203
            errors.append(e)
            continue
    raise FileNotFoundError(errors)

write(content)

Write the configuration source file and return the number of bytes written.

Parameters:

• content

  (AnyStr) –

  The content to write to the configuration source.

Source code in configzen/sources.py

def write(self, content: AnyStr) -> int:
    """
    Write the configuration source file and return the number of bytes written.

    Parameters
    ----------
    content
        The content to write to the configuration source.

    """
    if self.is_binary():
        return self.source.write_bytes(content)
    return self.source.write_text(content)

write_async(content)

Write the configuration source file asynchronously.

Return the number of bytes written.

Parameters:

• content

  (AnyStr) –

  The content to write to the configuration source.

Source code in configzen/sources.py

async def write_async(self, content: AnyStr) -> int:
    """
    Write the configuration source file asynchronously.

    Return the number of bytes written.

    Parameters
    ----------
    content
        The content to write to the configuration source.

    """
    if self.is_binary():
        return await AsyncPath(self.source).write_bytes(content)
    return await AsyncPath(self.source).write_text(content)

get_config_source(source, _data_format=None)

Get a dedicated interface for a configuration source.

Source code in configzen/sources.py

@singledispatch
def get_config_source(
    source: object,
    _data_format: DataFormat[Any, AnyStr] | None = None,
) -> ConfigSource[Any, Any]:
    """Get a dedicated interface for a configuration source."""
    type_name = type(source).__name__
    msg = (
        f"There is no class operating on {type_name!r} configuration "
        f"sources. Implement it by creating a subclass of ConfigSource."
    )
    raise NotImplementedError(msg)

get_stream_config_source(source, data_format)

Get a dedicated interface for a configuration source stream.

Source code in configzen/sources.py

@get_config_source.register(BytesIO)
@get_config_source.register(StringIO)
def get_stream_config_source(
    source: IO[bytes] | IO[str],
    data_format: DataFormat[Any, Any],
) -> StreamConfigSource[str] | StreamConfigSource[bytes]:
    """Get a dedicated interface for a configuration source stream."""
    return StreamConfigSource(source, data_format=data_format)

get_file_config_source(source, data_format=None)

Get a dedicated interface for a configuration source file.

Source code in configzen/sources.py

@get_config_source.register(str)
@get_config_source.register(bytes)
@get_config_source.register(PathLike)
def get_file_config_source(
    source: str | bytes | PathLike[str] | PathLike[bytes],
    data_format: DataFormat[Any, AnyStr] | None = None,
) -> FileConfigSource[str] | FileConfigSource[bytes]:
    """Get a dedicated interface for a configuration source file."""
    return FileConfigSource(source, data_format=data_format)

typedefs

Miscellaneous type definitions for configzen.