Add on_bind callback to dotenv parser for immediate env var visibility

Command substitutions in .env files now see variables defined earlier
in the same file by setting os.environ as each binding is parsed.

Author: davegaeddert

plain/plain/utils/dotenv.py

@@ -16,6 +16,7 @@
 import os
 import re
 import subprocess
+from collections.abc import Callable
 from pathlib import Path
 
 __all__ = ["load_dotenv", "parse_dotenv"]
@@ -46,13 +47,16 @@ def load_dotenv(
     if not path.exists():
         return False
 
+    content = path.read_text(encoding="utf-8")
+
     # Skip command execution for keys that already exist (unless override)
     skip_commands_for = None if override else set(os.environ.keys())
-    env_vars = _parse_dotenv_internal(path, skip_commands_for=skip_commands_for)
-    for key, value in env_vars.items():
+
+    def on_bind(key: str, value: str) -> None:
         if override or key not in os.environ:
             os.environ[key] = value
 
+    _parse_content(content, skip_commands_for=skip_commands_for, on_bind=on_bind)
     return True
 
 
@@ -62,27 +66,14 @@ def parse_dotenv(filepath: str | Path) -> dict[str, str]:
 
     Does not modify os.environ. Supports multiline values in quoted strings.
     """
-    return _parse_dotenv_internal(filepath, skip_commands_for=None)
-
-
-def _parse_dotenv_internal(
-    filepath: str | Path, skip_commands_for: set[str] | None = None
-) -> dict[str, str]:
-    """
-    Internal parser that can skip command execution for certain keys.
-
-    Args:
-        filepath: Path to the .env file
-        skip_commands_for: If provided, skip command substitution for keys in this set
-                          and use os.environ value instead
-    """
-    path = Path(filepath)
-    content = path.read_text(encoding="utf-8")
-    return _parse_content(content, skip_commands_for=skip_commands_for)
+    content = Path(filepath).read_text(encoding="utf-8")
+    return _parse_content(content)
 
 
 def _parse_content(
-    content: str, skip_commands_for: set[str] | None = None
+    content: str,
+    skip_commands_for: set[str] | None = None,
+    on_bind: Callable[[str, str], None] | None = None,
 ) -> dict[str, str]:
     """Parse .env file content and return key-value pairs."""
     result: dict[str, str] = {}
@@ -107,6 +98,8 @@ def _parse_content(
         if parsed:
             key, value, new_pos = parsed
             result[key] = value
+            if on_bind:
+                on_bind(key, value)
             pos = new_pos
         else:
             # Skip to next line on parse failure
@@ -136,7 +129,7 @@ def _parse_binding(
     length = len(content)
 
     # Skip optional 'export ' prefix
-    if content[pos:].startswith("export "):
+    if content[pos : pos + 7] == "export ":
         pos += 7
         while pos < length and content[pos] in " \t":
             pos += 1

plain/tests/test_dotenv.py

@@ -0,0 +1,82 @@
+import os
+
+import pytest
+
+from plain.utils.dotenv import load_dotenv, parse_dotenv
+
+
+@pytest.fixture
+def env_file(tmp_path):
+    """Helper to create a .env file with given content."""
+
+    def _create(content):
+        path = tmp_path / ".env"
+        path.write_text(content)
+        return path
+
+    return _create
+
+
+@pytest.fixture(autouse=True)
+def clean_env():
+    """Remove test keys from os.environ after each test."""
+    yield
+    for key in list(os.environ):
+        if key.startswith("TEST_"):
+            del os.environ[key]
+
+
+def test_command_substitution_sees_earlier_vars(env_file):
+    """Command substitution should see variables defined earlier in the same .env file."""
+    path = env_file("TEST_TOKEN=hello\nTEST_RESULT=$(echo $TEST_TOKEN)\n")
+    load_dotenv(path)
+    assert os.environ["TEST_TOKEN"] == "hello"
+    assert os.environ["TEST_RESULT"] == "hello"
+
+
+def test_command_substitution_chained(env_file):
+    """Multiple command substitutions can chain through os.environ."""
+    path = env_file(
+        "TEST_A=first\nTEST_B=$(echo $TEST_A)-second\nTEST_C=$(echo $TEST_B)\n"
+    )
+    load_dotenv(path)
+    assert os.environ["TEST_A"] == "first"
+    assert os.environ["TEST_B"] == "first-second"
+    assert os.environ["TEST_C"] == "first-second"
+
+
+def test_parse_dotenv_no_environ_side_effects(env_file):
+    """parse_dotenv should not modify os.environ."""
+    path = env_file("TEST_PARSE_ONLY=value\n")
+    result = parse_dotenv(path)
+    assert result == {"TEST_PARSE_ONLY": "value"}
+    assert "TEST_PARSE_ONLY" not in os.environ
+
+
+def test_load_dotenv_no_override_by_default(env_file):
+    """Existing env vars should not be overridden by default."""
+    os.environ["TEST_EXISTING"] = "original"
+    path = env_file("TEST_EXISTING=new_value\n")
+    load_dotenv(path)
+    assert os.environ["TEST_EXISTING"] == "original"
+
+
+def test_load_dotenv_override(env_file):
+    """With override=True, existing env vars should be replaced."""
+    os.environ["TEST_EXISTING"] = "original"
+    path = env_file("TEST_EXISTING=new_value\n")
+    load_dotenv(path, override=True)
+    assert os.environ["TEST_EXISTING"] == "new_value"
+
+
+def test_load_dotenv_missing_file(tmp_path):
+    """load_dotenv should return False for a missing file."""
+    result = load_dotenv(tmp_path / "nonexistent.env")
+    assert result is False
+
+
+def test_load_dotenv_returns_true(env_file):
+    """load_dotenv should return True when file exists."""
+    path = env_file("TEST_KEY=value\n")
+    result = load_dotenv(path)
+    assert result is True