Env parser (#42)

* add parser and few tests.

Author: HEROgold

src/confkit/data_types.py

@@ -101,7 +101,7 @@ def _strip_comment(value: str) -> str:
         Since hex values use 0x prefix (not #), we can safely strip everything after #.
         """
         if "#" in value:
-            return value.split("#")[0].strip()
+            return value.split("#", maxsplit=1)[0].strip()
         return value
 
     @abstractmethod

src/confkit/ext/parsers.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar
 
@@ -58,6 +59,98 @@ def get(self, section: str, option: str, fallback: str = UNSET) -> str:
     def set(self, section: str, option: str, value: str) -> None:
         """Set the value of an option within a section."""
 
+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."""

tests/test_env_parser.py

@@ -0,0 +1,181 @@
+"""Tests for EnvParser."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+
+import pytest
+
+from confkit.ext.parsers import EnvParser
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@pytest.fixture
+def env_parser() -> EnvParser:
+    """Create an EnvParser instance."""
+    return EnvParser()
+
+
+@pytest.fixture
+def temp_env_file(tmp_path: Path) -> Path:
+    """Create a temporary .env file."""
+    return tmp_path / ".env"
+
+
+def test_env_parser_init(env_parser: EnvParser) -> None:
+    """Test EnvParser initialization."""
+    assert isinstance(env_parser.data, dict)
+
+
+def test_env_parser_read_nonexistent_file(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test reading from a nonexistent file loads environment variables."""
+    # Set a test env var
+    test_key = "TEST_CONFKIT_VAR"
+    test_value = "test_value"
+    os.environ[test_key] = test_value
+
+    try:
+        env_parser.read(temp_env_file)
+        assert test_key in env_parser.data
+        assert env_parser.data[test_key] == test_value
+    finally:
+        del os.environ[test_key]
+
+
+def test_env_parser_read_env_file(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test reading from a .env file."""
+    # Create a .env file
+    temp_env_file.write_text("KEY1=value1\nKEY2=value2\nKEY3=value with spaces\n")
+
+    env_parser.read(temp_env_file)
+
+    assert "KEY1" in env_parser.data
+    assert env_parser.data["KEY1"] == "value1"
+    assert env_parser.data["KEY2"] == "value2"
+    assert env_parser.data["KEY3"] == "value with spaces"
+
+
+def test_env_parser_read_env_file_with_comments(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test reading .env file with comments and empty lines."""
+    content = """
+# This is a comment
+KEY1=value1
+
+# Another comment
+KEY2=value2
+    """
+    temp_env_file.write_text(content)
+
+    env_parser.read(temp_env_file)
+
+    assert "KEY1" in env_parser.data
+    assert "KEY2" in env_parser.data
+    assert env_parser.data["KEY1"] == "value1"
+    assert env_parser.data["KEY2"] == "value2"
+
+
+def test_env_parser_read_env_file_with_quotes(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test reading .env file with quoted values."""
+    content = 'KEY1="quoted value"\nKEY2=\'single quoted\'\nKEY3=unquoted\n'
+    temp_env_file.write_text(content)
+
+    env_parser.read(temp_env_file)
+
+    assert env_parser.data["KEY1"] == "quoted value"
+    assert env_parser.data["KEY2"] == "single quoted"
+    assert env_parser.data["KEY3"] == "unquoted"
+
+
+def test_env_parser_env_vars_take_precedence(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test that environment variables take precedence over .env file."""
+    # Create a .env file
+    temp_env_file.write_text("CONFKIT_TEST=from_file\n")
+
+    # Set environment variable with same key
+    os.environ["CONFKIT_TEST"] = "from_env"
+
+    try:
+        env_parser.read(temp_env_file)
+        assert env_parser.data["CONFKIT_TEST"] == "from_env"
+    finally:
+        del os.environ["CONFKIT_TEST"]
+
+
+def test_env_parser_write(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test that writing raises NotImplementedError for readonly parser."""
+    env_parser.data = {
+        "KEY1": "value1",
+        "KEY2": "value2",
+        "KEY3": "value with spaces",
+    }
+
+    with temp_env_file.open("w") as f, pytest.raises(NotImplementedError):
+        env_parser.write(f)
+
+
+def test_env_parser_has_section(env_parser: EnvParser) -> None:
+    """Test has_section always returns True."""
+    assert env_parser.has_section("any_section")
+    assert env_parser.has_section("")
+
+
+def test_env_parser_has_option(env_parser: EnvParser) -> None:
+    """Test has_option checks for key existence."""
+    env_parser.data = {"KEY1": "value1"}
+
+    assert env_parser.has_option("any_section", "KEY1")
+    assert not env_parser.has_option("any_section", "KEY2")
+
+
+def test_env_parser_get(env_parser: EnvParser) -> None:
+    """Test getting values."""
+    env_parser.data = {"KEY1": "value1"}
+
+    assert env_parser.get("any_section", "KEY1") == "value1"
+    assert env_parser.get("any_section", "KEY2") == ""
+    assert env_parser.get("any_section", "KEY2", fallback="default") == "default"
+
+
+def test_env_parser_set(env_parser: EnvParser) -> None:
+    """Test that setting values raises NotImplementedError for readonly parser."""
+    with pytest.raises(NotImplementedError):
+        env_parser.set("any_section", "KEY1", "value1")
+
+
+def test_env_parser_remove_option(env_parser: EnvParser) -> None:
+    """Test removing options."""
+    env_parser.data = {"KEY1": "value1", "KEY2": "value2"}
+
+    env_parser.remove_option("any_section", "KEY1")
+
+    assert "KEY1" not in env_parser.data
+    assert "KEY2" in env_parser.data
+
+
+def test_env_parser_add_section(env_parser: EnvParser) -> None:
+    """Test add_section is a no-op."""
+    env_parser.add_section("section")  # Should not raise
+
+
+def test_env_parser_set_section(env_parser: EnvParser) -> None:
+    """Test set_section is a no-op."""
+    env_parser.set_section("section")  # Should not raise
+
+
+def test_env_parser_integration(env_parser: EnvParser, temp_env_file: Path) -> None:
+    """Test full read cycle for readonly parser."""
+    # Create a .env file
+    content = """DATABASE_URL=postgresql://localhost/db
+API_KEY=secret123
+DEBUG=true"""
+    temp_env_file.write_text(content)
+
+    # Read from file
+    env_parser.read(temp_env_file)
+
+    assert env_parser.data["DATABASE_URL"] == "postgresql://localhost/db"
+    assert env_parser.data["API_KEY"] == "secret123"
+    assert env_parser.data["DEBUG"] == "true"