Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions .github/copilot-instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,10 @@ confkit is a Python library for type-safe configuration management using descrip
- `Config` descriptor (`config.py`): The main descriptor class that handles getting/setting values in config files
- `ConfigContainerMeta` (`config.py`): Metaclass that enables setting Config descriptors on class variables
- `BaseDataType` and implementations (`data_types.py`): Type converters for different data types
- `Parser` facade (`ext/parsers.py`): Unified facade for all configuration file formats (INI, JSON, YAML, TOML, .env)
- `IniParser` (`ext/parsers.py`): Adapter for Python's built-in ConfigParser (INI files)
- `ConfkitParser` protocol (`parsers.py`): Defines the unified parser interface for all configuration file formats (INI, JSON, YAML, TOML, .env)
- `IniParser` (`parsers.py`): Adapter for Python's built-in ConfigParser (INI files)
- `MsgspecParser` (`ext/parsers.py`): Adapter for JSON, YAML, and TOML files using msgspec
- `EnvParser` (`ext/parsers.py`): Adapter for environment variables and .env files
- `EnvParser` (`parsers.py`): Adapter for environment variables and .env files
- `sentinels.py`: Provides the `UNSET` sentinel value for representing unset values
- `exceptions.py`: Custom exceptions for configuration errors
- `watcher.py`: File watching functionality to detect config file changes
Expand Down
3 changes: 0 additions & 3 deletions examples/nested_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,6 @@
- TOML files (nested tables)
- INI files (using dot notation in section names)
"""
from confkit.ext.parsers import IniParser

from pathlib import Path
from typing import TypeVar

Expand Down Expand Up @@ -99,7 +97,6 @@ class IniConfig(Config[T]):
...


IniConfig.set_parser(IniParser())
IniConfig.set_file(Path("nested_example.ini"))


Expand Down
7 changes: 3 additions & 4 deletions src/confkit/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

from typing_extensions import deprecated

from confkit.ext.parsers import EnvParser, IniParser
from confkit.parsers import EnvParser, IniParser
from confkit.watcher import FileWatcher

from .data_types import BaseDataType, Optional
Expand All @@ -25,7 +25,7 @@
from collections.abc import Callable
from pathlib import Path

from confkit.ext.parsers import ConfkitParser
from confkit.parsers import ConfkitParser

# Type variables for Python 3.10+ (pre-PEP 695) compatibility
VT = TypeVar("VT")
Expand Down Expand Up @@ -311,8 +311,7 @@ def _set(cls, section: str, setting: str, value: VT | BaseDataType[VT] | BaseDat
if not cls._parser.has_section(section):
cls._parser.add_section(section)

sanitized_str = cls._sanitize_str(str(value))
cls._parser.set(section, setting, sanitized_str)
cls._parser.set(section, setting, value)

if cls.write_on_edit:
cls.write()
Expand Down
271 changes: 52 additions & 219 deletions src/confkit/ext/parsers.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,23 @@
"""Parsers for Confkit configuration files."""
"""Optional msgspec-based parsers for Confkit configuration files.

This module requires the ``msgspec`` optional extra:

pip install confkit[msgspec]
uv add confkit[msgspec]

Importing this module without ``msgspec`` installed will raise an
``ImportError`` immediately. Built-in parsers (``IniParser``,
``EnvParser``, ``ConfkitParser``) that have no optional dependencies
live in ``confkit.parsers``.
"""
from __future__ import annotations

import os
import sys
from configparser import ConfigParser
from io import TextIOWrapper
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar

from confkit.data_types import BaseDataType
from confkit.exceptions import ConfigPathConflictError
from confkit.parsers import ConfkitParser

try:
import msgspec
Expand All @@ -26,13 +35,13 @@


if sys.version_info >= (3, 12):
from typing import Protocol, override
from typing import override
# TD: Use nested types when Python 3.11 is EOL and we can drop support for it
# otherwise this gets syntax errors.
# type NestedDict = dict[str, NestedDict | str | int | float | bool | None] # noqa: ERA001
NestedDict = dict[str, Any]
else:
from typing_extensions import Protocol, override
from typing_extensions import override
NestedDict = dict[str, Any]

from confkit.sentinels import UNSET
Expand All @@ -45,173 +54,6 @@

T = TypeVar("T")

class ConfkitParser(Protocol):
"""A protocol for Confkit parsers."""

def read(self, file: Path) -> None:
"""Read the configuration from a file."""
def write(self, io: TextIOWrapper[_WrappedBuffer]) -> None:
"""Write the configuration to a file-like object."""
def has_section(self, section: str) -> bool:
"""Check if a section exists."""
def set_section(self, section: str) -> None:
"""Set a section."""
def set_option(self, option: str) -> None:
"""Set an option."""
def add_section(self, section: str) -> None:
"""Add a section."""
def has_option(self, section: str, option: str) -> bool:
"""Check if an option exists within a section."""
def remove_option(self, section: str, option: str) -> None:
"""Remove an option from a section."""
def get(self, section: str, option: str, fallback: str = UNSET) -> str:
"""Get the value of an option within a section, with an optional fallback."""
def set(self, section: str, option: str, value: str) -> None:
"""Set the value of an option within a section."""


class IniParser(ConfkitParser):
"""Adapter for ConfigParser that supports dot notation for nested sections."""

def __init__(self) -> None:
"""Initialize the IniParser with an internal ConfigParser instance."""
self.parser = ConfigParser()
self._file: Path | None = None

@override
def read(self, file: Path) -> None:
self.parser.read(file)

@override
def write(self, io: TextIOWrapper) -> None:
self.parser.write(io)

@override
def has_section(self, section: str) -> bool:
return self.parser.has_section(section)

@override
def set_section(self, section: str) -> None:
if not self.parser.has_section(section):
self.parser.add_section(section)

@override
def set_option(self, option: str) -> None:
# Not used directly; options are set via set()
pass

@override
def add_section(self, section: str) -> None:
self.parser.add_section(section)

@override
def has_option(self, section: str, option: str) -> bool:
return self.parser.has_option(section, option)

@override
def remove_option(self, section: str, option: str) -> None:
self.parser.remove_option(section, option)

@override
def get(self, section: str, option: str, fallback: str = UNSET) -> str:
return self.parser.get(section, option, fallback=fallback)

@override
def set(self, section: str, option: str, value: str) -> None:
self.parser.set(section, option, value)


class EnvParser(ConfkitParser):
"""A parser for environment variables and .env files.

This parser operates without sections - all configuration is stored as flat key-value pairs.
Values are read from environment variables and optionally persisted to a .env file.
"""

def __init__(self) -> None: # noqa: D107
self.data: dict[str, str] = {}

@override
def read(self, file: Path) -> None:
"""Precedence, from lowest to highest.

- config file
- environment vars
"""
self.data = dict(os.environ)

if not file.exists():
return

with file.open("r", encoding="utf-8") as f:
for i in f:
line = i.strip()
if not line or line.startswith("#"):
continue

match line.split("=", 1):
case [key, value]:
if key not in os.environ:
# Strip quotes from values
value = value.strip()
if (value.startswith('"') and value.endswith('"')) or \
(value.startswith("'") and value.endswith("'")):
value = value[1:-1]
self.data[key.strip()] = value

@override
def remove_option(self, section: str, option: str) -> None:
"""Remove an option (section is ignored)."""
if option in self.data:
del self.data[option]

@override
def get(self, section: str, option: str, fallback: str = UNSET) -> str:
"""Get the value of an option (section is ignored)."""
if option in self.data:
return self.data[option]
if fallback is not UNSET:
return str(fallback)
return ""

@override
def has_option(self, section: str, option: str) -> bool:
"""Check if an option exists (section is ignored)."""
return option in self.data

@override
def has_section(self, section: str) -> bool:
"""EnvParser has no sections, always returns True for compatibility."""
return True

@override
def write(self, io: TextIOWrapper[_WrappedBuffer]) -> None:
"""Write configuration to a .env file."""
msg = "EnvParser does not support writing to .env"
raise NotImplementedError(msg)

@override
def set_section(self, section: str) -> None:
"""EnvParser has no sections, this is a no-op."""
pass # noqa: PIE790

@override
def set_option(self, option: str) -> None:
"""Set an option (not used in EnvParser)."""
msg = "EnvParser does not support set_option"
raise NotImplementedError(msg)

@override
def add_section(self, section: str) -> None:
"""EnvParser has no sections, this is a no-op."""
pass # noqa: PIE790

@override
def set(self, section: str, option: str, value: str) -> None:
"""Set the value of an option (section is ignored)."""
msg = "EnvParser does not support set"
raise NotImplementedError(msg)


class MsgspecParser(ConfkitParser, Generic[T]):
"""Unified msgspec-based parser for YAML, JSON, TOML configuration files."""
Expand Down Expand Up @@ -266,15 +108,15 @@ def _navigate_to_section(self, section: str, *, create: bool = False) -> NestedD
Args:
section: Dot-separated section path (e.g., "Parent.Child.GrandChild")
create: If True, create missing intermediate sections and raise an error
if an intermediate path element is a scalar instead of a dict
if any path element is a scalar instead of a dict.

Returns:
The nested dict at the section path, or None if not found and create=False

Raises:
ConfigPathConflictError: When create=True and an intermediate path element
is a scalar value instead of a dict. This prevents
silent data loss from overwriting scalars.
ConfigPathConflictError: When create=True and any path element (including
the final one) is a scalar value instead of a dict.
This prevents silent data loss from overwriting scalars.

"""
if not section:
Expand All @@ -300,19 +142,26 @@ def _navigate_to_section(self, section: str, *, create: bool = False) -> NestedD
else:
return None
current = current[part]
# Check if we hit a scalar in the middle of the path
if i < len(parts) - 1 and not isinstance(current, dict):
# Check if we hit a scalar anywhere in the path (including final element when create=True)
if not isinstance(current, dict):
if create:
path_so_far = ".".join(parts[: i + 1])
msg = (
f"Cannot navigate to section '{section}': "
f"'{path_so_far}' is a scalar value, not a section. "
f"Path conflict at '{parts[i + 1]}'."
)
is_final = i == len(parts) - 1
if is_final:
msg = (
f"Cannot navigate to section '{section}': "
f"'{path_so_far}' is a scalar value, not a section."
)
else:
msg = (
f"Cannot navigate to section '{section}': "
f"'{path_so_far}' is a scalar value, not a section. "
f"Path conflict at '{parts[i + 1]}'."
)
raise ConfigPathConflictError(msg)
return None

return current if isinstance(current, dict) else None
return current # guaranteed to be a dict here

@override
def has_section(self, section: str) -> bool:
Expand All @@ -339,40 +188,24 @@ def get(self, section: str, option: str, fallback: str = UNSET) -> str:
return str(section_data[option])

@override
def set(self, section: str, option: str, value: str) -> None:
# This will raise ConfigPathConflictError if an intermediate path is a scalar
def set(self, section: str, option: str, value: object) -> None:
# Raises ConfigPathConflictError if any path element is a scalar.
section_data = self._navigate_to_section(section, create=True)
if section_data is not None:
# Try to preserve the original type by parsing the string value
# This is important for JSON/YAML/TOML which support native types
parsed_value = self._parse_value(value)
section_data[option] = parsed_value

def _parse_value(self, value: str) -> bool | int | float | str:
"""Parse a string value to its appropriate type for structured formats.

Attempts to convert string values back to their original types:
- "True"/"False" -> bool
- Integer strings -> int
- Float strings -> float
- Everything else remains a string
"""
if value == "True":
return True
if value == "False":
return False

try:
return int(value)
except ValueError:
pass

try:
return float(value)
except ValueError:
pass

return value
# _navigate_to_section always raises ConfigPathConflictError when create=True
# and the path is blocked, so section_data is guaranteed to be a dict here.
assert section_data is not None # noqa: S101
if isinstance(value, BaseDataType):
native = value.value
# BaseDataType.__str__ returns str(self.value) by default.
# Subclasses with custom string representations (e.g. Hex returns "0xa",
# Octal returns "0o10") override __str__, causing str(native) != str(value).
# In those cases, store the custom string so convert() can round-trip
# correctly on the next read. For standard types the native Python value
# is stored directly, preserving native JSON/YAML/TOML types.
stored = native if str(native) == str(value) else str(value)
else:
stored = value
section_data[option] = stored

@override
def remove_option(self, section: str, option: str) -> None:
Expand Down
Loading