From 002bfb667eb6bb3229a9b571d62dcd166491b2ad Mon Sep 17 00:00:00 2001 From: Akshat Talwar Date: Tue, 5 May 2026 21:32:02 +0530 Subject: [PATCH 1/2] Add on_duplicate parameter to handle duplicate keys Closes #591 Adds on_duplicate parameter to DotEnv, load_dotenv, and dotenv_values with three modes: - warn (default): logs a warning, latter value wins - raise: raises ValueError on first duplicate found - ignore: silently uses latter value, old behaviour explicit --- src/dotenv/main.py | 31 ++++++++++++++++ tests/test_on_duplicate.py | 73 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 tests/test_on_duplicate.py diff --git a/src/dotenv/main.py b/src/dotenv/main.py index d17b45cb..0887735c 100644 --- a/src/dotenv/main.py +++ b/src/dotenv/main.py @@ -12,6 +12,8 @@ from .parser import Binding, parse_stream from .variables import parse_variables +_DUPLICATE_VALUES = ("warn", "raise", "ignore") + # A type alias for a string path to be used for the paths in this file. # These paths may flow to `open()` and `os.replace()`. StrPath = Union[str, "os.PathLike[str]"] @@ -48,7 +50,13 @@ def __init__( encoding: Optional[str] = None, interpolate: bool = True, override: bool = True, + on_duplicate: str = "warn", ) -> None: + if on_duplicate not in _DUPLICATE_VALUES: + raise ValueError( + f"Invalid value for on_duplicate: {on_duplicate!r}. " + f"Expected one of: {', '.join(_DUPLICATE_VALUES)}" + ) self.dotenv_path: Optional[StrPath] = dotenv_path self.stream: Optional[IO[str]] = stream self._dict: Optional[Dict[str, Optional[str]]] = None @@ -56,6 +64,7 @@ def __init__( self.encoding: Optional[str] = encoding self.interpolate: bool = interpolate self.override: bool = override + self.on_duplicate: str = on_duplicate @contextmanager def _get_stream(self) -> Iterator[IO[str]]: @@ -90,8 +99,26 @@ def dict(self) -> Dict[str, Optional[str]]: def parse(self) -> Iterator[Tuple[str, Optional[str]]]: with self._get_stream() as stream: + seen_keys: Dict[str, int] = {} for mapping in with_warn_for_invalid_lines(parse_stream(stream)): if mapping.key is not None: + if mapping.key in seen_keys: + msg = ( + "Duplicate key %r found in %s " + "(first defined on line %d, redefined on line %d)." + ) + args = ( + mapping.key, + self.dotenv_path or "", + seen_keys[mapping.key], + mapping.original.line, + ) + if self.on_duplicate == "raise": + raise ValueError(msg % args) + elif self.on_duplicate == "warn": + logger.warning(msg, *args) + else: + seen_keys[mapping.key] = mapping.original.line yield mapping.key, mapping.value def set_as_environment_variables(self) -> bool: @@ -387,6 +414,7 @@ def load_dotenv( override: bool = False, interpolate: bool = True, encoding: Optional[str] = "utf-8", + on_duplicate: str = "warn", ) -> bool: """Parse a .env file and then load all the variables found as environment variables. @@ -426,6 +454,7 @@ def load_dotenv( interpolate=interpolate, override=override, encoding=encoding, + on_duplicate=on_duplicate, ) return dotenv.set_as_environment_variables() @@ -436,6 +465,7 @@ def dotenv_values( verbose: bool = False, interpolate: bool = True, encoding: Optional[str] = "utf-8", + on_duplicate: str = "warn", ) -> Dict[str, Optional[str]]: """ Parse a .env file and return its content as a dict. @@ -464,6 +494,7 @@ def dotenv_values( interpolate=interpolate, override=True, encoding=encoding, + on_duplicate=on_duplicate, ).dict() diff --git a/tests/test_on_duplicate.py b/tests/test_on_duplicate.py new file mode 100644 index 00000000..586de553 --- /dev/null +++ b/tests/test_on_duplicate.py @@ -0,0 +1,73 @@ +import logging +import os +from unittest.mock import patch + +import pytest + +import dotenv +from dotenv.main import DotEnv + + +def _write_env(tmp_path, content): + env_file = tmp_path / ".env" + env_file.write_text(content) + return env_file + + +class TestOnDuplicate: + def test_warn_emits_warning(self, tmp_path): + env_file = _write_env(tmp_path, "FOO=first\nBAR=ok\nFOO=second\n") + with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: + result = DotEnv(env_file, on_duplicate="warn").dict() + assert mock_warn.called + assert "Duplicate key" in mock_warn.call_args[0][0] + assert result["FOO"] == "second" + + def test_raise_raises_valueerror(self, tmp_path): + env_file = _write_env(tmp_path, "FOO=first\nFOO=second\n") + with pytest.raises(ValueError, match="Duplicate key"): + DotEnv(env_file, on_duplicate="raise").dict() + + def test_ignore_no_warning(self, tmp_path): + env_file = _write_env(tmp_path, "FOO=first\nFOO=second\n") + with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: + result = DotEnv(env_file, on_duplicate="ignore").dict() + assert not mock_warn.called + assert result["FOO"] == "second" + + def test_invalid_option_raises(self, tmp_path): + env_file = _write_env(tmp_path, "") + with pytest.raises(ValueError, match="Invalid value for on_duplicate"): + DotEnv(env_file, on_duplicate="bad-value") + + def test_load_dotenv_warn(self, tmp_path): + env_file = _write_env(tmp_path, "MYKEY=first\nMYKEY=second\n") + with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: + dotenv.load_dotenv(env_file, override=True, on_duplicate="warn") + assert mock_warn.called + assert "Duplicate key" in mock_warn.call_args[0][0] + del os.environ["MYKEY"] + + def test_load_dotenv_raise(self, tmp_path): + env_file = _write_env(tmp_path, "MYKEY=first\nMYKEY=second\n") + with pytest.raises(ValueError, match="Duplicate key"): + dotenv.load_dotenv(env_file, on_duplicate="raise") + + def test_dotenv_values_warn(self, tmp_path): + env_file = _write_env(tmp_path, "Z=1\nZ=2\n") + with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: + result = dotenv.dotenv_values(env_file, on_duplicate="warn") + assert mock_warn.called + assert result["Z"] == "2" + + def test_dotenv_values_raise(self, tmp_path): + env_file = _write_env(tmp_path, "Z=1\nZ=2\n") + with pytest.raises(ValueError, match="Duplicate key"): + dotenv.dotenv_values(env_file, on_duplicate="raise") + + def test_no_duplicate_no_warning(self, tmp_path): + env_file = _write_env(tmp_path, "A=1\nB=2\n") + with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: + result = dotenv.dotenv_values(env_file, on_duplicate="warn") + assert not mock_warn.called + assert result == {"A": "1", "B": "2"} From bb3516e1ecfd393fe0d9db42bf2449fb704392c8 Mon Sep 17 00:00:00 2001 From: Akshat Talwar Date: Tue, 5 May 2026 21:45:56 +0530 Subject: [PATCH 2/2] address review feedback: add docstrings, fix env cleanup in tests --- src/dotenv/main.py | 2 ++ tests/test_on_duplicate.py | 5 ++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/dotenv/main.py b/src/dotenv/main.py index 0887735c..d00dcc72 100644 --- a/src/dotenv/main.py +++ b/src/dotenv/main.py @@ -427,6 +427,8 @@ def load_dotenv( from the `.env` file. interpolate: Whether to interpolate variables using POSIX variable expansion. encoding: Encoding to be used to read the file. + on_duplicate: How to handle duplicate keys. "warn" logs a warning, + "raise" raises a ValueError, "ignore" silently uses the latter value. Returns: Bool: True if at least one environment variable is set else False diff --git a/tests/test_on_duplicate.py b/tests/test_on_duplicate.py index 586de553..e9359555 100644 --- a/tests/test_on_duplicate.py +++ b/tests/test_on_duplicate.py @@ -1,5 +1,4 @@ import logging -import os from unittest.mock import patch import pytest @@ -40,13 +39,13 @@ def test_invalid_option_raises(self, tmp_path): with pytest.raises(ValueError, match="Invalid value for on_duplicate"): DotEnv(env_file, on_duplicate="bad-value") - def test_load_dotenv_warn(self, tmp_path): + def test_load_dotenv_warn(self, tmp_path, monkeypatch): env_file = _write_env(tmp_path, "MYKEY=first\nMYKEY=second\n") + monkeypatch.delenv("MYKEY", raising=False) with patch.object(logging.getLogger("dotenv.main"), "warning") as mock_warn: dotenv.load_dotenv(env_file, override=True, on_duplicate="warn") assert mock_warn.called assert "Duplicate key" in mock_warn.call_args[0][0] - del os.environ["MYKEY"] def test_load_dotenv_raise(self, tmp_path): env_file = _write_env(tmp_path, "MYKEY=first\nMYKEY=second\n")