Add on_duplicate parameter to handle duplicate keys

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,14 +50,21 @@ 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
         self.verbose: bool = verbose
         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 "<stream>",
+                            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.
 
@@ -399,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
 
@@ -426,6 +456,7 @@ def load_dotenv(
         interpolate=interpolate,
         override=override,
         encoding=encoding,
+        on_duplicate=on_duplicate,
     )
     return dotenv.set_as_environment_variables()
 
@@ -436,6 +467,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 +496,7 @@ def dotenv_values(
         interpolate=interpolate,
         override=True,
         encoding=encoding,
+        on_duplicate=on_duplicate,
     ).dict()
 
 

tests/test_on_duplicate.py

@@ -0,0 +1,72 @@
+import logging
+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, 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]
+
+    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"}