feat: add wait_for_directory function to handle directory creation with timeout

Author: wakonig

bec_lib/bec_lib/file_utils.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import time
 import warnings
 from typing import TYPE_CHECKING
 
@@ -204,6 +205,27 @@ def get_full_path(
     return full_path
 
 
+def wait_for_directory(path: str, timeout: float = 10.0, interval: float = 0.1) -> None:
+    """
+    Wait for a directory to be created.
+
+    Args:
+        path (str): Path to the directory to wait for.
+        timeout (float, optional): Maximum time to wait in seconds. Defaults to 10.
+        interval (float, optional): Time to wait between checks in seconds. Defaults to 0.1.
+
+    Raises:
+        FileWriterError: If the timeout is reached before the directory is created.
+
+    """
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if os.path.isdir(path):
+            return
+        time.sleep(interval)
+    raise FileWriterError(f"Timeout reached while waiting for directory {path} to be created.")
+
+
 class FileWriter:
     """FileWriter for creating file paths and directories for services and devices."""
 

bec_lib/tests/test_file_utils.py

@@ -2,6 +2,8 @@
 
 # pylint: skip-file
 import os
+import threading
+import time
 from unittest import mock
 
 import pytest
@@ -15,6 +17,7 @@
     LogWriter,
     compile_file_components,
     get_full_path,
+    wait_for_directory,
 )
 from bec_lib.messages import ScanStatusMessage
 from bec_lib.tests.utils import ConnectorMock
@@ -259,6 +262,30 @@ def test_compile_file_components_valid_paths(kwargs, expected_path, description)
     assert file_path == expected_path, description
 
 
+def test_wait_for_directory_returns_when_directory_appears(tmpdir):
+    """wait_for_directory should stop polling once the directory exists."""
+    dir_path = tmpdir.join("created-later")
+
+    def _create_directory():
+        time.sleep(0.02)
+        dir_path.mkdir()
+
+    creator = threading.Thread(target=_create_directory)
+    creator.start()
+    try:
+        wait_for_directory(str(dir_path), timeout=1.0, interval=0.01)
+    finally:
+        creator.join()
+
+
+def test_wait_for_directory_raises_on_timeout(tmpdir):
+    """wait_for_directory should raise when the directory never appears."""
+    dir_path = tmpdir.join("never-created")
+
+    with pytest.raises(FileWriterError, match="Timeout reached while waiting for directory"):
+        wait_for_directory(str(dir_path), timeout=0.05, interval=0.01)
+
+
 @pytest.mark.parametrize(
     "scan_info",
     [