feat: add watch option to FileBrowser for automatic file change detection

maartenbreddels · claude · maartenbreddels · commit a0c75202a9e5 · 2026-01-13T11:07:33.000+01:00
The FileBrowser component now supports a `watch` parameter that, when enabled,
automatically monitors the current directory for file changes and refreshes
the file list.

This uses the same approach as the Style component's CSS hot reload feature,
using watchfiles.awatch() in an async task managed via use_effect.

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;
diff --git a/solara/components/file_browser.py b/solara/components/file_browser.py
@@ -1,3 +1,4 @@
+import asyncio
 import os
 from os.path import isfile, join
 from pathlib import Path
@@ -8,6 +9,11 @@
 import ipyvuetify as vy
 import traitlets
 
+try:
+    import watchfiles
+except ModuleNotFoundError:
+    watchfiles = None  # type: ignore
+
 import solara
 from solara.components import Div
 
@@ -84,6 +90,7 @@ def FileBrowser(
     on_file_name: Optional[Callable[[str], None]] = None,
     start_directory=None,
     can_select=False,
+    watch: bool = False,
 ):
     """File/directory browser at the server side.
 
@@ -109,6 +116,8 @@ def FileBrowser(
      * `directory_first`: If `True` directories are shown before files. Default: `False`.
      * `on_file_name`: (deprecated) Use on_file_open instead.
      * `start_directory`: (deprecated) Use directory instead.
+     * `watch`: If `True`, watch the current directory for file changes and automatically refresh the file list.
+        Requires the `watchfiles` package to be installed.
     """
     if start_directory is not None:
         directory = start_directory  # pragma: no cover
@@ -122,6 +131,8 @@ def FileBrowser(
     warning, set_warning = solara.use_state(cast(Optional[str], None))
     scroll_pos_stack, set_scroll_pos_stack = solara.use_state(cast(List[int], []))
     scroll_pos, set_scroll_pos = solara.use_state(0)
+    # Counter to trigger re-render when files change
+    file_change_counter, set_file_change_counter = solara.use_state(0)
     selected_private, set_selected_private = use_reactive_or_value(
         selected,
         on_value=on_path_select if can_select else lambda x: None,
@@ -143,6 +154,38 @@ def sync_directory_from_selected():
 
     solara.use_effect(sync_directory_from_selected, [selected_private])
 
+    def watch_directory():
+        if not watch:
+            return
+        if not watchfiles:
+            logger.warning("watchfiles not installed, cannot watch directory")
+            return
+
+        # Check if there's a running event loop before creating the coroutine
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            # No running event loop (e.g., in unit tests or non-server environments)
+            logger.warning("No running event loop, cannot watch directory for changes")
+            return
+
+        dir_to_watch = current_dir.value
+
+        async def watch_task():
+            try:
+                async for _ in watchfiles.awatch(dir_to_watch):
+                    logger.debug("Directory %s changed, refreshing file list", dir_to_watch)
+                    set_file_change_counter(lambda x: x + 1)
+            except RuntimeError:
+                pass  # swallow the RuntimeError: Already borrowed errors from watchfiles
+            except Exception:
+                logger.exception("Error watching directory")
+
+        future = asyncio.create_task(watch_task())
+        return future.cancel
+
+    solara.use_effect(watch_directory, [watch, current_dir.value])
+
     def change_dir(new_dir: Path):
         if os.access(new_dir, os.R_OK):
             current_dir.value = new_dir
diff --git a/tests/unit/file_browser_test.py b/tests/unit/file_browser_test.py
@@ -1,3 +1,4 @@
+import asyncio
 import platform
 import unittest.mock
 from pathlib import Path
@@ -290,3 +291,157 @@ def Test3():
     selected.value = current_dir.value / ".." / "unit" / ".."
     assert current_dir.value == HERE.parent
     rc.close()
+
+
+def test_file_browser_watch_no_event_loop(tmpdir: Path):
+    """Test that watch=True gracefully handles no running event loop."""
+    tmpdir = Path(tmpdir)
+    (tmpdir / "test.txt").write_text("test")
+
+    # When watch=True but no event loop is available, it should log a warning but not crash
+    with unittest.mock.patch.object(solara.components.file_browser.logger, "warning") as mock_warning:
+
+        @solara.component
+        def Test():
+            return solara.FileBrowser(tmpdir, watch=True)
+
+        div, rc = solara.render_fixed(Test(), handle_error=False)
+        file_list: solara.components.file_browser.FileListWidget = div.children[1]
+
+        # Verify files are still shown correctly
+        files = {k["name"] for k in file_list.files}
+        assert "test.txt" in files
+
+        # Verify warning was logged about no event loop
+        mock_warning.assert_called_with("No running event loop, cannot watch directory for changes")
+        rc.close()
+
+
+def test_file_browser_watch_disabled_by_default():
+    """Test that watch is disabled by default (no warning when watchfiles not available)."""
+
+    @solara.component
+    def Test():
+        return solara.FileBrowser(HERE.parent)
+
+    # This should work without any issues even if watchfiles is not installed
+    div, rc = solara.render_fixed(Test(), handle_error=False)
+    file_list: solara.components.file_browser.FileListWidget = div.children[1]
+    assert "file_browser_test.py" in file_list
+    rc.close()
+
+
+def test_file_browser_watch_no_watchfiles(tmpdir: Path):
+    """Test that watch=True logs a warning when watchfiles is not installed."""
+    tmpdir = Path(tmpdir)
+    (tmpdir / "test.txt").write_text("test")
+
+    with unittest.mock.patch.object(solara.components.file_browser, "watchfiles", None):
+        with unittest.mock.patch.object(solara.components.file_browser.logger, "warning") as mock_warning:
+
+            @solara.component
+            def Test():
+                return solara.FileBrowser(tmpdir, watch=True)
+
+            div, rc = solara.render_fixed(Test(), handle_error=False)
+
+            # Give the effect a chance to run
+            import time
+
+            time.sleep(0.1)
+
+            mock_warning.assert_called_with("watchfiles not installed, cannot watch directory")
+            rc.close()
+
+
+@pytest.mark.asyncio
+async def test_file_browser_watch_detects_new_file(tmpdir: Path):
+    """Test that watch=True actually detects when a new file is added."""
+    tmpdir = Path(tmpdir)
+    (tmpdir / "initial.txt").write_text("initial")
+
+    files_changed_event = asyncio.Event()
+
+    @solara.component
+    def Test():
+        return solara.FileBrowser(tmpdir, watch=True)
+
+    div, rc = solara.render_fixed(Test(), handle_error=False)
+    file_list: solara.components.file_browser.FileListWidget = div.children[1]
+
+    # Verify initial state
+    initial_files = {k["name"] for k in file_list.files}
+    assert "initial.txt" in initial_files
+    assert "new_file.txt" not in initial_files
+
+    # Set up observer to detect when files trait changes
+    def on_files_change(change):
+        files_changed_event.set()
+
+    file_list.observe(on_files_change, "files")
+
+    # Give the watcher task a moment to start
+    await asyncio.sleep(0.1)
+
+    # Create a new file - this should trigger the watcher
+    (tmpdir / "new_file.txt").write_text("new content")
+
+    # Wait for the watcher to detect the change
+    try:
+        await asyncio.wait_for(files_changed_event.wait(), timeout=5.0)
+    except asyncio.TimeoutError:
+        pytest.fail("File change was not detected within timeout")
+
+    # Verify the new file is now in the list
+    new_files = {k["name"] for k in file_list.files}
+    assert "new_file.txt" in new_files
+    assert "initial.txt" in new_files
+
+    rc.close()
+
+
+@pytest.mark.asyncio
+async def test_file_browser_watch_detects_deleted_file(tmpdir: Path):
+    """Test that watch=True detects when a file is deleted."""
+    tmpdir = Path(tmpdir)
+    (tmpdir / "file1.txt").write_text("content1")
+    (tmpdir / "file2.txt").write_text("content2")
+
+    files_changed_event = asyncio.Event()
+
+    @solara.component
+    def Test():
+        return solara.FileBrowser(tmpdir, watch=True)
+
+    div, rc = solara.render_fixed(Test(), handle_error=False)
+    file_list: solara.components.file_browser.FileListWidget = div.children[1]
+
+    # Verify initial state
+    initial_files = {k["name"] for k in file_list.files}
+    assert "file1.txt" in initial_files
+    assert "file2.txt" in initial_files
+
+    # Set up observer
+    def on_files_change(change):
+        files_changed_event.set()
+
+    file_list.observe(on_files_change, "files")
+
+    # Give the watcher task a moment to start
+    await asyncio.sleep(0.1)
+
+    # Delete a file
+    (tmpdir / "file2.txt").unlink()
+
+    # Wait for the watcher to detect the change
+    try:
+        await asyncio.wait_for(files_changed_event.wait(), timeout=5.0)
+    except asyncio.TimeoutError:
+        pytest.fail("File deletion was not detected within timeout")
+
+    # Verify file2 is gone
+    new_files = {k["name"] for k in file_list.files}
+    assert "file1.txt" in new_files
+    assert "file2.txt" not in new_files
+
+    rc.close()
