From d6ec6bd9eabbe1d7a372527d911c99547bec76ee Mon Sep 17 00:00:00 2001 From: Jannik Reinhard Date: Sun, 26 Jul 2026 16:34:53 +0200 Subject: [PATCH] fix staging file processing lifecycle --- .github/workflows/ci.yml | 2 +- app/database.py | 3 +- app/main.py | 40 ++++++------------- app/services/file_watcher.py | 17 ++++---- tests/test_file_watcher.py | 77 ++++++++++++++++++++++++++++++++++++ 5 files changed, 101 insertions(+), 38 deletions(-) create mode 100644 tests/test_file_watcher.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cbf75af..2f2282a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,7 +19,7 @@ jobs: python-version: "3.12" cache: pip - name: Install test dependencies - run: pip install fastapi pydantic-settings sqlalchemy loguru pytest + run: pip install fastapi pydantic-settings sqlalchemy loguru watchdog pytest - name: Run regression tests run: python -m pytest tests -q - name: Parse PowerShell scripts diff --git a/app/database.py b/app/database.py index 0652cae..5441ea1 100644 --- a/app/database.py +++ b/app/database.py @@ -1,6 +1,5 @@ from sqlalchemy import create_engine -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.orm import sessionmaker, Session +from sqlalchemy.orm import declarative_base, sessionmaker, Session from loguru import logger import os from .config import get_settings diff --git a/app/main.py b/app/main.py index 84f5768..6ef3629 100644 --- a/app/main.py +++ b/app/main.py @@ -154,33 +154,19 @@ async def ensure_defaults(): async def start_file_watcher(): await asyncio.sleep(3) # Wait for database initialization try: - from .services.file_watcher import FileWatcherHandler - from .services.document_processor import DocumentProcessor - from watchdog.observers import Observer - - db = SessionLocal() - try: - settings = get_settings(db) - staging_path = Path(settings.staging_folder) - - # Ensure staging folder exists - staging_path.mkdir(parents=True, exist_ok=True) - - # Create document processor and file watcher - processor = DocumentProcessor(db) - event_handler = FileWatcherHandler(processor, settings, db) - - # Set up observer - global file_watcher - file_watcher = Observer() - file_watcher.schedule(event_handler, str(staging_path), recursive=False) - file_watcher.start() - - print(f"📁 File watcher started monitoring: {staging_path}") - - finally: - db.close() - + from .services.file_watcher import FileWatcher + + watcher = FileWatcher() + # FileWatcher.start also recovers files that arrived before the + # observer was ready. Run the initial scan off the event loop + # because OCR can be CPU intensive. + await asyncio.to_thread(watcher.start) + if not watcher.is_running: + raise RuntimeError("File watcher did not start") + + global file_watcher + file_watcher = watcher + print(f"📁 File watcher started monitoring: {watcher.settings.staging_folder}") except Exception as e: print(f"⚠️ Could not start file watcher: {e}") diff --git a/app/services/file_watcher.py b/app/services/file_watcher.py index f67bd8d..0db38a0 100644 --- a/app/services/file_watcher.py +++ b/app/services/file_watcher.py @@ -14,8 +14,7 @@ class FileWatcherHandler(FileSystemEventHandler): """Handler for file system events""" - def __init__(self, processor: DocumentProcessor, settings, db: Session = None): - self.processor = processor + def __init__(self, settings): self.settings = settings # Track recently processed files to prevent duplicates self._recent_files = defaultdict(datetime) @@ -59,10 +58,13 @@ def on_created(self, event): logger.info(f"New file detected: {file_path.name}") - # Process the file in background + # Create the processor with the same live session used for this job. + # Keeping a startup-time processor leaked closed SQLAlchemy sessions + # into later AI/settings operations. try: with SessionLocal() as db: - result = self.processor.process_file(file_path, db) + processor = DocumentProcessor(db) + result = processor.process_file(file_path, db) if result: logger.info(f"Successfully processed file: {file_path.name}") else: @@ -90,7 +92,6 @@ class FileWatcher: def __init__(self, db: Session = None): self.settings = None self.observer = Observer() - self.processor = None self.is_running = False self._initialized = False @@ -100,7 +101,6 @@ def _ensure_initialized(self): try: with SessionLocal() as db: self.settings = get_settings(db) - self.processor = DocumentProcessor(db) self._initialized = True except Exception as e: logger.error(f"Failed to initialize file watcher: {e}") @@ -115,7 +115,7 @@ def start(self): logger.warning(f"Staging folder does not exist: {staging_path}") return - handler = FileWatcherHandler(self.processor, self.settings) + handler = FileWatcherHandler(self.settings) self.observer.schedule(handler, str(staging_path), recursive=False) try: @@ -157,7 +157,8 @@ def _process_existing_files(self): try: with SessionLocal() as db: - self.processor.process_file(file_path, db) + processor = DocumentProcessor(db) + processor.process_file(file_path, db) except Exception as e: logger.error(f"Failed to process existing file {file_path.name}: {e}") diff --git a/tests/test_file_watcher.py b/tests/test_file_watcher.py new file mode 100644 index 0000000..275c0a0 --- /dev/null +++ b/tests/test_file_watcher.py @@ -0,0 +1,77 @@ +import importlib +import sys +import types +from pathlib import Path +from types import SimpleNamespace + + +class FakeSession: + def __init__(self): + self.active = False + + def __enter__(self): + self.active = True + return self + + def __exit__(self, *_args): + self.active = False + + +def load_file_watcher(monkeypatch): + processed = [] + + class FakeProcessor: + def __init__(self, db): + assert db.active + self.db = db + + def process_file(self, path, db): + assert db is self.db + assert db.active + processed.append(path) + return object() + + processor_module = types.ModuleType("app.services.document_processor") + processor_module.DocumentProcessor = FakeProcessor + monkeypatch.setitem(sys.modules, "app.services.document_processor", processor_module) + + sys.modules.pop("app.services.file_watcher", None) + module = importlib.import_module("app.services.file_watcher") + monkeypatch.setattr(module, "SessionLocal", FakeSession) + monkeypatch.setattr(module.time, "sleep", lambda _seconds: None) + return module, processed + + +def test_created_file_uses_a_live_database_session(tmp_path, monkeypatch): + module, processed = load_file_watcher(monkeypatch) + uploaded_file = tmp_path / "invoice.pdf" + uploaded_file.write_bytes(b"document") + settings = SimpleNamespace(allowed_extensions_list=["pdf"]) + handler = module.FileWatcherHandler(settings) + + handler.on_created(SimpleNamespace(is_directory=False, src_path=str(uploaded_file))) + + assert processed == [uploaded_file] + + +def test_start_recovers_files_that_precede_the_observer(tmp_path, monkeypatch): + module, _processed = load_file_watcher(monkeypatch) + + class FakeObserver: + def schedule(self, *_args, **_kwargs): + pass + + def start(self): + pass + + watcher = module.FileWatcher() + watcher.observer = FakeObserver() + watcher.settings = SimpleNamespace(staging_folder=str(tmp_path), allowed_extensions_list=["pdf"]) + watcher._initialized = True + recovered = [] + monkeypatch.setattr(watcher, "_process_existing_files", lambda: recovered.append(True)) + + watcher.start() + + assert watcher.is_running is True + assert recovered == [True]