From fe9ddfc7cad02eb83143f7dca1a1d9548c51bf4a Mon Sep 17 00:00:00 2001 From: Sidharth Sudhir Date: Tue, 21 Oct 2025 18:50:37 -0400 Subject: [PATCH 1/2] Support reading .env from FIFOs (Unix) --- src/dotenv/main.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/dotenv/main.py b/src/dotenv/main.py index b6de171c..7480b80c 100644 --- a/src/dotenv/main.py +++ b/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 os.path.isfile(self.dotenv_path) or stat.S_ISFIFO(os.stat(self.dotenv_path).st_mode): 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 os.path.isfile(check_path) or stat.S_ISFIFO(os.stat(check_path).st_mode): return check_path if raise_error_if_not_found: From 9a32042f227f38e76a55445210ed33c7725dd1df Mon Sep 17 00:00:00 2001 From: Sidharth Sudhir Date: Tue, 21 Oct 2025 19:20:49 -0400 Subject: [PATCH 2/2] handle FileNotFoundError in FIFO checks and add FIFO load test --- src/dotenv/main.py | 19 +++++++++++++++++-- tests/test_fifo_dotenv.py | 33 +++++++++++++++++++++++++++++++++ tests/test_main.py | 18 +++++++++++++----- 3 files changed, 63 insertions(+), 7 deletions(-) create mode 100644 tests/test_fifo_dotenv.py diff --git a/src/dotenv/main.py b/src/dotenv/main.py index 7480b80c..1d6bf0b0 100644 --- a/src/dotenv/main.py +++ b/src/dotenv/main.py @@ -62,7 +62,7 @@ def __init__( @contextmanager def _get_stream(self) -> Iterator[IO[str]]: - if self.dotenv_path and os.path.isfile(self.dotenv_path) or stat.S_ISFIFO(os.stat(self.dotenv_path).st_mode): + 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: @@ -326,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) or stat.S_ISFIFO(os.stat(check_path).st_mode): + if _is_file_or_fifo(check_path): return check_path if raise_error_if_not_found: @@ -418,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) diff --git a/tests/test_fifo_dotenv.py b/tests/test_fifo_dotenv.py new file mode 100644 index 00000000..4961adce --- /dev/null +++ b/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" diff --git a/tests/test_main.py b/tests/test_main.py index 08b41cd3..44961117 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -263,7 +263,9 @@ def test_load_dotenv_existing_file(dotenv_path): ) def test_load_dotenv_disabled(dotenv_path, flag_value): expected_environ = {"PYTHON_DOTENV_DISABLED": flag_value} - with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True): + with mock.patch.dict( + os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True + ): dotenv_path.write_text("a=b") result = dotenv.load_dotenv(dotenv_path) @@ -289,7 +291,9 @@ def test_load_dotenv_disabled(dotenv_path, flag_value): ], ) def test_load_dotenv_disabled_notification(dotenv_path, flag_value): - with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True): + with mock.patch.dict( + os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True + ): dotenv_path.write_text("a=b") logger = logging.getLogger("dotenv.main") @@ -298,7 +302,7 @@ def test_load_dotenv_disabled_notification(dotenv_path, flag_value): assert result is False mock_debug.assert_called_once_with( - "python-dotenv: .env loading disabled by PYTHON_DOTENV_DISABLED environment variable" + "python-dotenv: .env loading disabled by PYTHON_DOTENV_DISABLED environment variable" ) @@ -321,7 +325,9 @@ def test_load_dotenv_disabled_notification(dotenv_path, flag_value): ) def test_load_dotenv_enabled(dotenv_path, flag_value): expected_environ = {"PYTHON_DOTENV_DISABLED": flag_value, "a": "b"} - with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True): + with mock.patch.dict( + os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True + ): dotenv_path.write_text("a=b") result = dotenv.load_dotenv(dotenv_path) @@ -348,7 +354,9 @@ def test_load_dotenv_enabled(dotenv_path, flag_value): ], ) def test_load_dotenv_enabled_no_notification(dotenv_path, flag_value): - with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True): + with mock.patch.dict( + os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True + ): dotenv_path.write_text("a=b") logger = logging.getLogger("dotenv.main")