Support reading .env from FIFOs (Unix) (#586)

* Support reading .env from FIFOs (Unix)

* handle FileNotFoundError in FIFO checks and add FIFO load test

---------

Co-authored-by: Sidharth Sudhir <[email protected]>

Author: sidharth-sudhir

src/dotenv/main.py

@@ -3,6 +3,7 @@
 import os
 import pathlib
 import shutil
+import stat
 import sys
 import tempfile
 from collections import OrderedDict
@@ -61,7 +62,7 @@ def __init__(
 
     @contextmanager
     def _get_stream(self) -> Iterator[IO[str]]:
-        if self.dotenv_path and os.path.isfile(self.dotenv_path):
+        if self.dotenv_path and _is_file_or_fifo(self.dotenv_path):
             with open(self.dotenv_path, encoding=self.encoding) as stream:
                 yield stream
         elif self.stream is not None:
@@ -325,7 +326,7 @@ def _is_debugger():
 
     for dirname in _walk_to_root(path):
         check_path = os.path.join(dirname, filename)
-        if os.path.isfile(check_path):
+        if _is_file_or_fifo(check_path):
             return check_path
 
     if raise_error_if_not_found:
@@ -417,3 +418,18 @@ def dotenv_values(
         override=True,
         encoding=encoding,
     ).dict()
+
+
+def _is_file_or_fifo(path: StrPath) -> bool:
+    """
+    Return True if `path` exists and is either a regular file or a FIFO.
+    """
+    if os.path.isfile(path):
+        return True
+
+    try:
+        st = os.stat(path)
+    except (FileNotFoundError, OSError):
+        return False
+
+    return stat.S_ISFIFO(st.st_mode)

tests/test_fifo_dotenv.py

@@ -0,0 +1,33 @@
+import os
+import pathlib
+import sys
+import threading
+
+import pytest
+
+from dotenv import load_dotenv
+
+pytestmark = pytest.mark.skipif(
+    sys.platform.startswith("win"), reason="FIFOs are Unix-only"
+)
+
+
+def test_load_dotenv_from_fifo(tmp_path: pathlib.Path, monkeypatch):
+    fifo = tmp_path / ".env"
+    os.mkfifo(fifo)  # create named pipe
+
+    def writer():
+        with open(fifo, "w", encoding="utf-8") as w:
+            w.write("MY_PASSWORD=pipe-secret\n")
+
+    t = threading.Thread(target=writer)
+    t.start()
+
+    # Ensure env is clean
+    monkeypatch.delenv("MY_PASSWORD", raising=False)
+
+    ok = load_dotenv(dotenv_path=str(fifo), override=True)
+    t.join(timeout=2)
+
+    assert ok is True
+    assert os.getenv("MY_PASSWORD") == "pipe-secret"