feat: Add waiting logic for FIFO files that may not be immediately readable

src/dotenv/main.py

@@ -2,10 +2,12 @@
 import logging
 import os
 import pathlib
+import select
 import shutil
 import stat
 import sys
 import tempfile
+import time
 from collections import OrderedDict
 from contextlib import contextmanager
 from typing import IO, Dict, Iterable, Iterator, Mapping, Optional, Tuple, Union
@@ -63,7 +65,11 @@ def __init__(
     @contextmanager
     def _get_stream(self) -> Iterator[IO[str]]:
         if self.dotenv_path and _is_file_or_fifo(self.dotenv_path):
-            with open(self.dotenv_path, encoding=self.encoding) as stream:
+            # Handle files that may need to wait for content to become available
+            # This includes FIFOs (named pipes) and mounted filesystems that may
+            # initially be empty or not yet populated when first accessed
+            stream = _wait_for_file_content(self.dotenv_path, encoding=self.encoding)
+            with stream:
                 yield stream
         elif self.stream is not None:
             yield self.stream
@@ -420,6 +426,48 @@ def dotenv_values(
     ).dict()
 
 
+def _wait_for_file_content(
+    path: StrPath,
+    encoding: Optional[str] = None,
+    max_wait_time: float = 5.0,
+) -> IO[str]:
+    """
+    Wait for file content to be available, handling both regular files and pipes (FIFOs).
+
+    Some environments expose .env data via FIFOs; reading the pipe produces content on demand.
+    For FIFOs we block until the pipe is readable, then read once and return a StringIO over
+    the decoded bytes. For regular files we open and return the handle directly.
+    """
+    start_time = time.time()
+
+    try:
+        st = os.stat(path)
+        is_fifo = stat.S_ISFIFO(st.st_mode)
+    except (FileNotFoundError, OSError):
+        is_fifo = False
+
+    if not is_fifo:
+        # Regular file path: open once and return immediately
+        return open(path, encoding=encoding)
+
+    # FIFO path: block until readable (up to max_wait_time), then read once.
+    # Open unbuffered binary so select() reflects readiness accurately before decoding.
+    with open(path, "rb", buffering=0) as fifo:
+        fd = fifo.fileno()
+        timeout = max_wait_time - (time.time() - start_time)
+        if timeout < 0:
+            timeout = 0
+        ready, _, _ = select.select([fd], [], [], timeout)
+        if not ready:
+            # If it never became readable, return empty content for caller to handle
+            return io.StringIO("")
+
+        raw = fifo.read()
+        text = raw.decode(encoding or "utf-8", errors="replace")
+        # Return a fresh StringIO so caller can read from the start
+        return io.StringIO(text)
+
+
 def _is_file_or_fifo(path: StrPath) -> bool:
     """
     Return True if `path` exists and is either a regular file or a FIFO.

tests/test_fifo_dotenv.py

@@ -6,6 +6,7 @@
 import pytest
 
 from dotenv import load_dotenv
+from dotenv.main import _wait_for_file_content
 
 pytestmark = pytest.mark.skipif(
     sys.platform.startswith("win"), reason="FIFOs are Unix-only"
@@ -31,3 +32,35 @@ def writer():
 
     assert ok is True
     assert os.getenv("MY_PASSWORD") == "pipe-secret"
+
+
+def test_wait_for_file_content_with_regular_file(tmp_path: pathlib.Path):
+    """Test that _wait_for_file_content handles regular files correctly."""
+    regular_file = tmp_path / ".env"
+    regular_file.write_text("KEY=value\n", encoding="utf-8")
+
+    stream = _wait_for_file_content(str(regular_file), encoding="utf-8")
+    content = stream.read()
+
+    assert content == "KEY=value\n"
+    stream.close()
+
+
+def test_wait_for_file_content_with_fifo(tmp_path: pathlib.Path):
+    """Test that _wait_for_file_content handles FIFOs correctly."""
+    fifo = tmp_path / ".env"
+    os.mkfifo(fifo)
+
+    def writer():
+        with open(fifo, "w", encoding="utf-8") as w:
+            w.write("FIFO_KEY=fifo-value\n")
+
+    t = threading.Thread(target=writer)
+    t.start()
+
+    stream = _wait_for_file_content(str(fifo), encoding="utf-8")
+    content = stream.read()
+
+    t.join(timeout=2)
+
+    assert content == "FIFO_KEY=fifo-value\n"