From 3743ed4ca8d01534ab4449709aa1881363299bf6 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 31 Mar 2026 10:34:07 +0200 Subject: [PATCH] Refactor backend to hexagonal architecture for converter extensibility Extract domain value objects and ports from parsing.py, move Docling-specific code to infra/local_converter.py, and convert analysis_service to a class with injected DocumentConverter. This prepares the codebase for plugging in alternative conversion backends (e.g. Docling Serve) via the Protocol pattern. Co-Authored-By: Claude Opus 4.6 --- document-parser/api/analyses.py | 15 +- document-parser/domain/parsing.py | 310 ++---------------- document-parser/domain/ports.py | 23 ++ document-parser/domain/value_objects.py | 51 +++ document-parser/infra/__init__.py | 0 document-parser/infra/local_converter.py | 243 ++++++++++++++ document-parser/infra/settings.py | 30 ++ document-parser/main.py | 63 +++- document-parser/services/analysis_service.py | 155 ++++----- document-parser/tests/test_api_endpoints.py | 22 +- .../tests/test_pipeline_options.py | 117 +++---- 11 files changed, 579 insertions(+), 450 deletions(-) create mode 100644 document-parser/domain/ports.py create mode 100644 document-parser/domain/value_objects.py create mode 100644 document-parser/infra/__init__.py create mode 100644 document-parser/infra/local_converter.py create mode 100644 document-parser/infra/settings.py diff --git a/document-parser/api/analyses.py b/document-parser/api/analyses.py index 3d91805..3a978fe 100644 --- a/document-parser/api/analyses.py +++ b/document-parser/api/analyses.py @@ -7,12 +7,17 @@ import logging from fastapi import APIRouter, HTTPException from api.schemas import AnalysisResponse, CreateAnalysisRequest -from services import analysis_service logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/analyses", tags=["analyses"]) +def _get_service(): + """Lazy import to avoid circular dependency at module level.""" + from main import analysis_service + return analysis_service + + def _to_response(job) -> AnalysisResponse: return AnalysisResponse( id=job.id, @@ -40,7 +45,7 @@ async def create_analysis(body: CreateAnalysisRequest): pipeline_opts = body.pipelineOptions.model_dump() try: - job = await analysis_service.create(body.documentId, pipeline_options=pipeline_opts) + job = await _get_service().create(body.documentId, pipeline_options=pipeline_opts) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) from e @@ -50,14 +55,14 @@ async def create_analysis(body: CreateAnalysisRequest): @router.get("", response_model=list[AnalysisResponse]) async def list_analyses(): """List all analysis jobs.""" - jobs = await analysis_service.find_all() + jobs = await _get_service().find_all() return [_to_response(j) for j in jobs] @router.get("/{job_id}", response_model=AnalysisResponse) async def get_analysis(job_id: str): """Get a single analysis job.""" - job = await analysis_service.find_by_id(job_id) + job = await _get_service().find_by_id(job_id) if not job: raise HTTPException(status_code=404, detail="Analysis not found") return _to_response(job) @@ -66,6 +71,6 @@ async def get_analysis(job_id: str): @router.delete("/{job_id}", status_code=204) async def delete_analysis(job_id: str): """Delete an analysis job.""" - deleted = await analysis_service.delete(job_id) + deleted = await _get_service().delete(job_id) if not deleted: raise HTTPException(status_code=404, detail="Analysis not found") diff --git a/document-parser/domain/parsing.py b/document-parser/domain/parsing.py index 7e7b4bb..4a44e94 100644 --- a/document-parser/domain/parsing.py +++ b/document-parser/domain/parsing.py @@ -1,298 +1,34 @@ -"""Docling document extraction logic — pure domain, no HTTP concerns. +"""Backward-compatible re-exports for domain.parsing. -Wraps the Docling library to convert documents and extract structured -per-page elements with bounding boxes and hierarchy levels. +After the hexagonal architecture refactoring: +- Value objects moved to domain.value_objects +- Docling implementation moved to infra.local_converter + +This module re-exports the public names so existing code and tests +that import from domain.parsing continue to work. """ from __future__ import annotations -import contextlib -import logging -import threading -from dataclasses import dataclass, field - -from docling.datamodel.base_models import InputFormat -from docling.datamodel.pipeline_options import ( - PdfPipelineOptions, - TableFormerMode, - TableStructureOptions, +from domain.value_objects import ( # noqa: F401 + ConversionOptions, + ConversionResult, + PageDetail, + PageElement, ) -from docling.document_converter import DocumentConverter, PdfFormatOption -from docling_core.types.doc import ( - CodeItem, - DocItem, - FloatingItem, - FormulaItem, - GroupItem, - ListItem, - PictureItem, - SectionHeaderItem, - TableItem, - TextItem, - TitleItem, +from infra.local_converter import ( + _build_docling_converter, + _convert_sync, + _extract_pages_detail as extract_pages_detail, # noqa: F401 + _get_default_converter as get_default_converter, # noqa: F401 ) -from domain.bbox import to_topleft_list -logger = logging.getLogger(__name__) - -# Thread lock — DocumentConverter is not thread-safe -_converter_lock = threading.Lock() - -# US Letter page dimensions (points) — fallback when page size is unknown -_DEFAULT_PAGE_WIDTH = 612.0 -_DEFAULT_PAGE_HEIGHT = 792.0 - -# Default converter (lazy-init on first request) -_default_converter: DocumentConverter | None = None +def build_converter(options: ConversionOptions | None = None): + """Build a Docling DocumentConverter (backward-compatible signature).""" + return _build_docling_converter(options or ConversionOptions()) -# --------------------------------------------------------------------------- -# Domain value objects -# --------------------------------------------------------------------------- - -@dataclass -class PageElement: - type: str - bbox: list[float] - content: str - level: int = 0 - - -@dataclass -class PageDetail: - page_number: int - width: float - height: float - elements: list[PageElement] = field(default_factory=list) - - -@dataclass -class ConversionOptions: - do_ocr: bool = True - do_table_structure: bool = True - table_mode: str = "accurate" - do_code_enrichment: bool = False - do_formula_enrichment: bool = False - do_picture_classification: bool = False - do_picture_description: bool = False - generate_picture_images: bool = False - generate_page_images: bool = False - images_scale: float = 1.0 - - def is_default(self) -> bool: - return self == ConversionOptions() - - -@dataclass -class ConversionResult: - page_count: int - content_markdown: str - content_html: str - pages: list[PageDetail] - skipped_items: int = 0 - - -# --------------------------------------------------------------------------- -# Element type detection -# --------------------------------------------------------------------------- - -# Mapping from Docling type to element type string. -# Order matters: most specific types before their parents. -_ELEMENT_TYPE_MAP: list[tuple[type, str]] = [ - (TableItem, "table"), - (PictureItem, "picture"), - (TitleItem, "title"), - (SectionHeaderItem, "section_header"), - (ListItem, "list"), - (FormulaItem, "formula"), - (CodeItem, "code"), - (FloatingItem, "floating"), - (TextItem, "text"), -] - - -def _get_element_type(item: DocItem) -> str: - """Determine element type via isinstance on Docling's type hierarchy.""" - for cls, type_name in _ELEMENT_TYPE_MAP: - if isinstance(item, cls): - return type_name - return "text" - - -# --------------------------------------------------------------------------- -# Pipeline factory -# --------------------------------------------------------------------------- - -def build_converter(options: ConversionOptions | None = None) -> DocumentConverter: - """Build a DocumentConverter with the given pipeline options.""" - opts = options or ConversionOptions() - - table_options = TableStructureOptions( - do_cell_matching=True, - mode=TableFormerMode.ACCURATE if opts.table_mode == "accurate" else TableFormerMode.FAST, - ) - - pipeline_options = PdfPipelineOptions( - do_ocr=opts.do_ocr, - do_table_structure=opts.do_table_structure, - table_structure_options=table_options, - do_code_enrichment=opts.do_code_enrichment, - do_formula_enrichment=opts.do_formula_enrichment, - do_picture_classification=opts.do_picture_classification, - do_picture_description=opts.do_picture_description, - generate_page_images=opts.generate_page_images, - generate_picture_images=opts.generate_picture_images, - images_scale=opts.images_scale, - ) - - return DocumentConverter( - format_options={ - InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options), - } - ) - - -def get_default_converter() -> DocumentConverter: - global _default_converter - if _default_converter is None: - _default_converter = build_converter() - return _default_converter - - -# --------------------------------------------------------------------------- -# Page extraction -# --------------------------------------------------------------------------- - -def extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]: - """Extract per-page element details with bounding boxes from Docling result. - - Returns (pages, skipped_count) for transparent error reporting. - """ - pages: dict[int, PageDetail] = {} - document = doc_result.document - skipped = 0 - - for page_key, page_obj in document.pages.items(): - page_no = int(page_key) if isinstance(page_key, str) else page_key - pages[page_no] = PageDetail( - page_number=page_no, - width=page_obj.size.width, - height=page_obj.size.height, - ) - - for item, level in document.iterate_items(): - ok = _process_content_item(item, level, pages) - if not ok: - skipped += 1 - - sorted_pages = sorted(pages.values(), key=lambda p: p.page_number) - return sorted_pages, skipped - - -def _process_content_item( - item: DocItem | GroupItem, level: int, pages: dict[int, PageDetail], -) -> bool: - """Process a single content item and add it to the appropriate page.""" - if isinstance(item, GroupItem): - return True - - if not isinstance(item, DocItem) or not item.prov: - return False - - for prov in item.prov: - try: - page_no = prov.page_no - if page_no not in pages: - # Fallback: page was not found in document.pages (corrupted PDF or - # Docling edge case). US Letter dimensions are used as a safe default. - # This may cause slight bbox misalignment on non-Letter pages (e.g. A4). - logger.warning( - "Page %d not found in document metadata — using US Letter fallback (%sx%s pt)", - page_no, _DEFAULT_PAGE_WIDTH, _DEFAULT_PAGE_HEIGHT, - ) - pages[page_no] = PageDetail(page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT) - - page_height = pages[page_no].height - - bbox = [0.0, 0.0, 0.0, 0.0] - if prov.bbox: - bbox = to_topleft_list(prov.bbox, page_height) - - element_type = _get_element_type(item) - - content = getattr(item, "text", "") or "" - if isinstance(item, TableItem): - with contextlib.suppress(AttributeError, ValueError): - content = item.export_to_markdown() - - pages[page_no].elements.append( - PageElement(type=element_type, bbox=bbox, content=content, level=level) - ) - except (AttributeError, KeyError, TypeError, ValueError): - logger.warning( - "Skipping item %s on page %s", - type(item).__name__, - getattr(prov, "page_no", "?"), - exc_info=True, - ) - return False - - return True - - -# --------------------------------------------------------------------------- -# Main conversion entry point -# --------------------------------------------------------------------------- - -def _select_converter(options: ConversionOptions) -> DocumentConverter: - """Return the cached default converter or build a custom one.""" - if options.is_default(): - return get_default_converter() - return build_converter(options) - - -def _build_fallback_pages(doc, page_count: int) -> list[PageDetail]: - """Create empty PageDetail entries when extraction yields nothing.""" - return [ - PageDetail( - page_number=i + 1, - width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH, - height=doc.pages[i + 1].size.height if (i + 1) in doc.pages else _DEFAULT_PAGE_HEIGHT, - ) - for i in range(page_count) - ] - - -def convert_document( - file_path: str, - options: ConversionOptions | None = None, -) -> ConversionResult: - """Convert a document and return structured results. - - This is the main entry point for document parsing. Runs synchronously - (caller should use asyncio.to_thread for non-blocking execution). - """ - opts = options or ConversionOptions() - - with _converter_lock: - conv = _select_converter(opts) - result = conv.convert(file_path) - - doc = result.document - page_count = len(doc.pages) - pages_detail, skipped = extract_pages_detail(result) - - if not pages_detail and page_count > 0: - pages_detail = _build_fallback_pages(doc, page_count) - - if skipped > 0: - logger.info("Parsed: %d pages, %d items skipped", page_count, skipped) - - return ConversionResult( - page_count=page_count or len(pages_detail) or 1, - content_markdown=doc.export_to_markdown(), - content_html=doc.export_to_html(), - pages=pages_detail, - skipped_items=skipped, - ) +def convert_document(file_path: str, options: ConversionOptions | None = None) -> ConversionResult: + """Convert a document synchronously (backward-compatible signature).""" + return _convert_sync(file_path, options or ConversionOptions()) diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py new file mode 100644 index 0000000..562c567 --- /dev/null +++ b/document-parser/domain/ports.py @@ -0,0 +1,23 @@ +"""Domain ports — abstract interfaces that infrastructure must implement. + +These protocols define what the domain NEEDS, not how it's done. +Infrastructure adapters (local Docling, Docling Serve, etc.) implement these. +""" + +from __future__ import annotations + +from typing import Protocol + +from domain.value_objects import ConversionOptions, ConversionResult + + +class DocumentConverter(Protocol): + """Port for document conversion. + + Any implementation (local Docling lib, remote Docling Serve, mock, etc.) + must satisfy this contract. + """ + + async def convert( + self, file_path: str, options: ConversionOptions, + ) -> ConversionResult: ... diff --git a/document-parser/domain/value_objects.py b/document-parser/domain/value_objects.py new file mode 100644 index 0000000..b4c5959 --- /dev/null +++ b/document-parser/domain/value_objects.py @@ -0,0 +1,51 @@ +"""Domain value objects — pure data structures for document conversion. + +These types define the contract between the domain and infrastructure layers. +They have ZERO external dependencies (no docling, no HTTP, no DB). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class PageElement: + type: str + bbox: list[float] + content: str + level: int = 0 + + +@dataclass +class PageDetail: + page_number: int + width: float + height: float + elements: list[PageElement] = field(default_factory=list) + + +@dataclass +class ConversionOptions: + do_ocr: bool = True + do_table_structure: bool = True + table_mode: str = "accurate" + do_code_enrichment: bool = False + do_formula_enrichment: bool = False + do_picture_classification: bool = False + do_picture_description: bool = False + generate_picture_images: bool = False + generate_page_images: bool = False + images_scale: float = 1.0 + + def is_default(self) -> bool: + return self == ConversionOptions() + + +@dataclass +class ConversionResult: + page_count: int + content_markdown: str + content_html: str + pages: list[PageDetail] + skipped_items: int = 0 diff --git a/document-parser/infra/__init__.py b/document-parser/infra/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py new file mode 100644 index 0000000..efd4a5a --- /dev/null +++ b/document-parser/infra/local_converter.py @@ -0,0 +1,243 @@ +"""Local Docling converter — runs Docling as a Python library in-process. + +This adapter implements the DocumentConverter port using the Docling library +directly. It wraps the blocking DocumentConverter in asyncio.to_thread for +non-blocking execution. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import threading + +from docling.datamodel.base_models import InputFormat +from docling.datamodel.pipeline_options import ( + PdfPipelineOptions, + TableFormerMode, + TableStructureOptions, +) +from docling.document_converter import DocumentConverter as DoclingConverter +from docling.document_converter import PdfFormatOption +from docling_core.types.doc import ( + CodeItem, + DocItem, + FloatingItem, + FormulaItem, + GroupItem, + ListItem, + PictureItem, + SectionHeaderItem, + TableItem, + TextItem, + TitleItem, +) + +from domain.bbox import to_topleft_list +from domain.value_objects import ( + ConversionOptions, + ConversionResult, + PageDetail, + PageElement, +) + +logger = logging.getLogger(__name__) + +# Thread lock — DoclingConverter is not thread-safe +_converter_lock = threading.Lock() + +# US Letter page dimensions (points) — fallback when page size is unknown +_DEFAULT_PAGE_WIDTH = 612.0 +_DEFAULT_PAGE_HEIGHT = 792.0 + +# Default converter (lazy-init on first request) +_default_converter: DoclingConverter | None = None + + +# --------------------------------------------------------------------------- +# Element type detection +# --------------------------------------------------------------------------- + +_ELEMENT_TYPE_MAP: list[tuple[type, str]] = [ + (TableItem, "table"), + (PictureItem, "picture"), + (TitleItem, "title"), + (SectionHeaderItem, "section_header"), + (ListItem, "list"), + (FormulaItem, "formula"), + (CodeItem, "code"), + (FloatingItem, "floating"), + (TextItem, "text"), +] + + +def _get_element_type(item: DocItem) -> str: + for cls, type_name in _ELEMENT_TYPE_MAP: + if isinstance(item, cls): + return type_name + return "text" + + +# --------------------------------------------------------------------------- +# Pipeline factory +# --------------------------------------------------------------------------- + +def _build_docling_converter(options: ConversionOptions) -> DoclingConverter: + table_options = TableStructureOptions( + do_cell_matching=True, + mode=TableFormerMode.ACCURATE if options.table_mode == "accurate" else TableFormerMode.FAST, + ) + + pipeline_options = PdfPipelineOptions( + do_ocr=options.do_ocr, + do_table_structure=options.do_table_structure, + table_structure_options=table_options, + do_code_enrichment=options.do_code_enrichment, + do_formula_enrichment=options.do_formula_enrichment, + do_picture_classification=options.do_picture_classification, + do_picture_description=options.do_picture_description, + generate_page_images=options.generate_page_images, + generate_picture_images=options.generate_picture_images, + images_scale=options.images_scale, + ) + + return DoclingConverter( + format_options={ + InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options), + } + ) + + +def _get_default_converter() -> DoclingConverter: + global _default_converter + if _default_converter is None: + _default_converter = _build_docling_converter(ConversionOptions()) + return _default_converter + + +def _select_converter(options: ConversionOptions) -> DoclingConverter: + if options.is_default(): + return _get_default_converter() + return _build_docling_converter(options) + + +# --------------------------------------------------------------------------- +# Page extraction +# --------------------------------------------------------------------------- + +def _extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]: + pages: dict[int, PageDetail] = {} + document = doc_result.document + skipped = 0 + + for page_key, page_obj in document.pages.items(): + page_no = int(page_key) if isinstance(page_key, str) else page_key + pages[page_no] = PageDetail( + page_number=page_no, + width=page_obj.size.width, + height=page_obj.size.height, + ) + + for item, level in document.iterate_items(): + ok = _process_content_item(item, level, pages) + if not ok: + skipped += 1 + + sorted_pages = sorted(pages.values(), key=lambda p: p.page_number) + return sorted_pages, skipped + + +def _process_content_item( + item: DocItem | GroupItem, level: int, pages: dict[int, PageDetail], +) -> bool: + if isinstance(item, GroupItem): + return True + + if not isinstance(item, DocItem) or not item.prov: + return False + + for prov in item.prov: + try: + page_no = prov.page_no + if page_no not in pages: + logger.warning( + "Page %d not found in document metadata — using US Letter fallback (%sx%s pt)", + page_no, _DEFAULT_PAGE_WIDTH, _DEFAULT_PAGE_HEIGHT, + ) + pages[page_no] = PageDetail(page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT) + + page_height = pages[page_no].height + + bbox = [0.0, 0.0, 0.0, 0.0] + if prov.bbox: + bbox = to_topleft_list(prov.bbox, page_height) + + element_type = _get_element_type(item) + + content = getattr(item, "text", "") or "" + if isinstance(item, TableItem): + with contextlib.suppress(AttributeError, ValueError): + content = item.export_to_markdown() + + pages[page_no].elements.append( + PageElement(type=element_type, bbox=bbox, content=content, level=level) + ) + except (AttributeError, KeyError, TypeError, ValueError): + logger.warning( + "Skipping item %s on page %s", + type(item).__name__, + getattr(prov, "page_no", "?"), + exc_info=True, + ) + return False + + return True + + +# --------------------------------------------------------------------------- +# Synchronous conversion (called via asyncio.to_thread) +# --------------------------------------------------------------------------- + +def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: + with _converter_lock: + conv = _select_converter(options) + result = conv.convert(file_path) + + doc = result.document + page_count = len(doc.pages) + pages_detail, skipped = _extract_pages_detail(result) + + if not pages_detail and page_count > 0: + pages_detail = [ + PageDetail( + page_number=i + 1, + width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH, + height=doc.pages[i + 1].size.height if (i + 1) in doc.pages else _DEFAULT_PAGE_HEIGHT, + ) + for i in range(page_count) + ] + + if skipped > 0: + logger.info("Parsed: %d pages, %d items skipped", page_count, skipped) + + return ConversionResult( + page_count=page_count or len(pages_detail) or 1, + content_markdown=doc.export_to_markdown(), + content_html=doc.export_to_html(), + pages=pages_detail, + skipped_items=skipped, + ) + + +# --------------------------------------------------------------------------- +# Public adapter class +# --------------------------------------------------------------------------- + +class LocalConverter: + """Adapter that runs Docling locally as a Python library.""" + + async def convert( + self, file_path: str, options: ConversionOptions, + ) -> ConversionResult: + return await asyncio.to_thread(_convert_sync, file_path, options) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py new file mode 100644 index 0000000..0b19c54 --- /dev/null +++ b/document-parser/infra/settings.py @@ -0,0 +1,30 @@ +"""Centralized application settings — loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Settings: + conversion_engine: str = "local" # "local" or "remote" + docling_serve_url: str = "http://localhost:5001" + docling_serve_api_key: str | None = None + conversion_timeout: int = 600 + upload_dir: str = "./uploads" + db_path: str = "./data/docling_studio.db" + cors_origins: list[str] = field(default_factory=lambda: ["http://localhost:3000", "http://localhost:5173"]) + + @classmethod + def from_env(cls) -> Settings: + cors_raw = os.environ.get("CORS_ORIGINS", "http://localhost:3000,http://localhost:5173") + return cls( + conversion_engine=os.environ.get("CONVERSION_ENGINE", "local"), + docling_serve_url=os.environ.get("DOCLING_SERVE_URL", "http://localhost:5001"), + docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"), + conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "600")), + upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), + db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"), + cors_origins=[o.strip() for o in cors_raw.split(",")], + ) diff --git a/document-parser/main.py b/document-parser/main.py index 7e29776..fda1072 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -1,14 +1,17 @@ """Docling Studio — unified FastAPI backend. -Single service replacing both the Spring Boot backend and the document parser. -Provides document management (upload, CRUD), analysis orchestration (async Docling -processing), and PDF preview — all backed by SQLite. +Single service providing document management (upload, CRUD), analysis +orchestration (async Docling processing), and PDF preview — all backed +by SQLite. + +Conversion engine is selected via CONVERSION_ENGINE env var: +- "local" → Docling runs in-process as a Python library (default) +- "remote" → delegates to a Docling Serve instance via HTTP """ from __future__ import annotations import logging -import os from contextlib import asynccontextmanager from fastapi import FastAPI @@ -16,6 +19,7 @@ from fastapi.middleware.cors import CORSMiddleware from api.analyses import router as analyses_router from api.documents import router as documents_router +from infra.settings import Settings from persistence.database import init_db logging.basicConfig( @@ -24,12 +28,49 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Settings & dependency wiring +# --------------------------------------------------------------------------- + +settings = Settings.from_env() + + +def _build_converter(): + """Build the converter adapter based on configuration.""" + if settings.conversion_engine == "remote": + from infra.serve_converter import ServeConverter + logger.info("Using remote Docling Serve at %s", settings.docling_serve_url) + return ServeConverter( + base_url=settings.docling_serve_url, + api_key=settings.docling_serve_api_key, + ) + else: + from infra.local_converter import LocalConverter + logger.info("Using local Docling converter") + return LocalConverter() + + +def _build_analysis_service(): + from services.analysis_service import AnalysisService + converter = _build_converter() + return AnalysisService( + converter=converter, + conversion_timeout=settings.conversion_timeout, + ) + + +# Singleton service instance — imported by API routers +analysis_service = _build_analysis_service() + + +# --------------------------------------------------------------------------- +# FastAPI app +# --------------------------------------------------------------------------- @asynccontextmanager async def lifespan(app: FastAPI): - """Startup: initialize database. Shutdown: nothing special needed.""" await init_db() - logger.info("Docling Studio backend ready") + logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine) yield @@ -39,20 +80,14 @@ app = FastAPI( lifespan=lifespan, ) -# CORS — configurable via env, defaults for local dev -allowed_origins = os.environ.get( - "CORS_ORIGINS", "http://localhost:3000,http://localhost:5173" -).split(",") - app.add_middleware( CORSMiddleware, - allow_origins=[o.strip() for o in allowed_origins], + allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "DELETE", "OPTIONS"], allow_headers=["Content-Type", "Authorization"], ) -# Mount routers app.include_router(documents_router) app.include_router(analyses_router) @@ -60,4 +95,4 @@ app.include_router(analyses_router) @app.get("/health") def health(): """Health check endpoint.""" - return {"status": "ok"} + return {"status": "ok", "engine": settings.conversion_engine} diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 7a4d71a..8759ef3 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -1,4 +1,8 @@ -"""Analysis service — async document parsing orchestration.""" +"""Analysis service — async document parsing orchestration. + +Uses an injected DocumentConverter (port) so the service is decoupled +from the conversion implementation (local Docling lib vs remote Docling Serve). +""" from __future__ import annotations @@ -8,32 +12,88 @@ import logging from dataclasses import asdict from domain.models import AnalysisJob -from domain.parsing import ConversionOptions, ConversionResult, convert_document +from domain.ports import DocumentConverter +from domain.value_objects import ConversionOptions, ConversionResult from persistence import analysis_repo, document_repo logger = logging.getLogger(__name__) -# Maximum time (seconds) allowed for a single document conversion. -CONVERSION_TIMEOUT = int(__import__("os").environ.get("CONVERSION_TIMEOUT", "600")) +class AnalysisService: + """Orchestrates document analysis using an injected converter.""" -async def create(document_id: str, *, pipeline_options: dict | None = None) -> AnalysisJob: - """Create a new analysis job and launch background processing.""" - doc = await document_repo.find_by_id(document_id) - if not doc: - raise ValueError(f"Document not found: {document_id}") + def __init__(self, converter: DocumentConverter, conversion_timeout: int = 600): + self._converter = converter + self._conversion_timeout = conversion_timeout - job = AnalysisJob(document_id=document_id) - job.document_filename = doc.filename - await analysis_repo.insert(job) + async def create(self, document_id: str, *, pipeline_options: dict | None = None) -> AnalysisJob: + """Create a new analysis job and launch background processing.""" + doc = await document_repo.find_by_id(document_id) + if not doc: + raise ValueError(f"Document not found: {document_id}") - # Fire background task with error logging callback - task = asyncio.create_task( - _run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options) - ) - task.add_done_callback(_on_task_done) + job = AnalysisJob(document_id=document_id) + job.document_filename = doc.filename + await analysis_repo.insert(job) - return job + task = asyncio.create_task( + self._run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options) + ) + task.add_done_callback(_on_task_done) + + return job + + async def find_all(self) -> list[AnalysisJob]: + return await analysis_repo.find_all() + + async def find_by_id(self, job_id: str) -> AnalysisJob | None: + return await analysis_repo.find_by_id(job_id) + + async def delete(self, job_id: str) -> bool: + return await analysis_repo.delete(job_id) + + async def _run_analysis( + self, job_id: str, file_path: str, filename: str, pipeline_options: dict | None = None, + ) -> None: + """Background task: run conversion and update job status.""" + try: + job = await analysis_repo.find_by_id(job_id) + if not job: + logger.error("Analysis job %s not found", job_id) + return + + job.mark_running() + await analysis_repo.update_status(job) + logger.info("Analysis started: %s (file: %s)", job_id, filename) + + options = ConversionOptions(**(pipeline_options or {})) + + result: ConversionResult = await asyncio.wait_for( + self._converter.convert(file_path, options), + timeout=self._conversion_timeout, + ) + + pages_json = json.dumps([asdict(p) for p in result.pages]) + + job.mark_completed( + markdown=result.content_markdown, + html=result.content_html, + pages_json=pages_json, + ) + await analysis_repo.update_status(job) + + if result.page_count: + await document_repo.update_page_count(job.document_id, result.page_count) + + logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count) + + except TimeoutError: + logger.error("Analysis timed out after %ds: %s", self._conversion_timeout, job_id) + await _mark_failed(job_id, f"Conversion timed out after {self._conversion_timeout}s") + + except Exception as e: + logger.exception("Analysis failed: %s", job_id) + await _mark_failed(job_id, str(e)) def _on_task_done(task: asyncio.Task) -> None: @@ -46,65 +106,6 @@ def _on_task_done(task: asyncio.Task) -> None: logger.error("Unhandled exception in analysis task: %s", exc, exc_info=True) -async def find_all() -> list[AnalysisJob]: - return await analysis_repo.find_all() - - -async def find_by_id(job_id: str) -> AnalysisJob | None: - return await analysis_repo.find_by_id(job_id) - - -async def delete(job_id: str) -> bool: - return await analysis_repo.delete(job_id) - - -async def _run_analysis( - job_id: str, file_path: str, filename: str, pipeline_options: dict | None = None, -) -> None: - """Background task: run Docling conversion and update job status.""" - try: - job = await analysis_repo.find_by_id(job_id) - if not job: - logger.error("Analysis job %s not found", job_id) - return - - job.mark_running() - await analysis_repo.update_status(job) - logger.info("Analysis started: %s (file: %s)", job_id, filename) - - # Build conversion options from pipeline dict - options = ConversionOptions(**(pipeline_options or {})) - - # Run blocking Docling conversion in a thread with timeout - result: ConversionResult = await asyncio.wait_for( - asyncio.to_thread(convert_document, file_path, options), - timeout=CONVERSION_TIMEOUT, - ) - - pages_json = json.dumps([asdict(p) for p in result.pages]) - - job.mark_completed( - markdown=result.content_markdown, - html=result.content_html, - pages_json=pages_json, - ) - await analysis_repo.update_status(job) - - # Update document page count if available - if result.page_count: - await document_repo.update_page_count(job.document_id, result.page_count) - - logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count) - - except TimeoutError: - logger.error("Analysis timed out after %ds: %s", CONVERSION_TIMEOUT, job_id) - await _mark_failed(job_id, f"Conversion timed out after {CONVERSION_TIMEOUT}s") - - except Exception as e: - logger.exception("Analysis failed: %s", job_id) - await _mark_failed(job_id, str(e)) - - async def _mark_failed(job_id: str, error: str) -> None: """Safely mark a job as failed, handling DB errors gracefully.""" try: diff --git a/document-parser/tests/test_api_endpoints.py b/document-parser/tests/test_api_endpoints.py index da10af9..d7970cb 100644 --- a/document-parser/tests/test_api_endpoints.py +++ b/document-parser/tests/test_api_endpoints.py @@ -18,7 +18,9 @@ class TestHealthEndpoint: def test_health(self, client): resp = client.get("/health") assert resp.status_code == 200 - assert resp.json() == {"status": "ok"} + data = resp.json() + assert data["status"] == "ok" + assert "engine" in data class TestDocumentEndpoints: @@ -102,7 +104,7 @@ class TestDocumentEndpoints: class TestAnalysisEndpoints: - @patch("services.analysis_service.find_all", new_callable=AsyncMock) + @patch("main.analysis_service.find_all", new_callable=AsyncMock) def test_list_analyses(self, mock_find_all, client): mock_find_all.return_value = [ AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf"), @@ -117,7 +119,7 @@ class TestAnalysisEndpoints: assert data[0]["documentFilename"] == "test.pdf" assert data[0]["status"] == "PENDING" - @patch("services.analysis_service.find_by_id", new_callable=AsyncMock) + @patch("main.analysis_service.find_by_id", new_callable=AsyncMock) def test_get_analysis(self, mock_find, client): job = AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf") job.mark_running() @@ -129,14 +131,14 @@ class TestAnalysisEndpoints: assert data["status"] == "RUNNING" assert data["startedAt"] is not None - @patch("services.analysis_service.find_by_id", new_callable=AsyncMock) + @patch("main.analysis_service.find_by_id", new_callable=AsyncMock) def test_get_analysis_not_found(self, mock_find, client): mock_find.return_value = None resp = client.get("/api/analyses/missing") assert resp.status_code == 404 - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_create_analysis(self, mock_create, client): mock_create.return_value = AnalysisJob( id="j1", document_id="d1", document_filename="test.pdf", @@ -149,7 +151,7 @@ class TestAnalysisEndpoints: assert data["documentId"] == "d1" mock_create.assert_called_once_with("d1", pipeline_options=None) - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_create_analysis_with_pipeline_options(self, mock_create, client): mock_create.return_value = AnalysisJob( id="j2", document_id="d1", document_filename="test.pdf", @@ -182,7 +184,7 @@ class TestAnalysisEndpoints: assert opts["generate_picture_images"] is True assert opts["images_scale"] == 2.0 - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_create_analysis_with_partial_pipeline_options(self, mock_create, client): """Pipeline options should use defaults for unspecified fields.""" mock_create.return_value = AnalysisJob( @@ -202,7 +204,7 @@ class TestAnalysisEndpoints: assert opts["table_mode"] == "accurate" assert opts["do_code_enrichment"] is False - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_create_analysis_document_not_found(self, mock_create, client): mock_create.side_effect = ValueError("Document not found: d99") @@ -217,14 +219,14 @@ class TestAnalysisEndpoints: resp = client.post("/api/analyses", json={"documentId": " "}) assert resp.status_code == 400 - @patch("services.analysis_service.delete", new_callable=AsyncMock) + @patch("main.analysis_service.delete", new_callable=AsyncMock) def test_delete_analysis(self, mock_delete, client): mock_delete.return_value = True resp = client.delete("/api/analyses/j1") assert resp.status_code == 204 - @patch("services.analysis_service.delete", new_callable=AsyncMock) + @patch("main.analysis_service.delete", new_callable=AsyncMock) def test_delete_analysis_not_found(self, mock_delete, client): mock_delete.return_value = False diff --git a/document-parser/tests/test_pipeline_options.py b/document-parser/tests/test_pipeline_options.py index dd48a86..41509c5 100644 --- a/document-parser/tests/test_pipeline_options.py +++ b/document-parser/tests/test_pipeline_options.py @@ -131,8 +131,8 @@ class TestBuildConverter: class TestConvertDocumentRouting: """Verify convert_document uses default converter for default opts, custom otherwise.""" - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_default_converter_with_all_defaults(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -148,8 +148,8 @@ class TestConvertDocumentRouting: mock_get_default.assert_called_once() mock_build.assert_not_called() - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_ocr_disabled(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -165,8 +165,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once() mock_get_default.assert_not_called() - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_table_mode_fast(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -182,8 +182,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once_with(opts) - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_code_enrichment_on(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -199,8 +199,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once_with(opts) - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_formula_enrichment_on(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -215,8 +215,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once() - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_picture_options_on(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -231,8 +231,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once() - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_generate_images_on(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -247,8 +247,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once() - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_uses_custom_converter_when_images_scale_changed(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -264,8 +264,8 @@ class TestConvertDocumentRouting: mock_build.assert_called_once_with(opts) - @patch("domain.parsing.get_default_converter") - @patch("domain.parsing.build_converter") + @patch("infra.local_converter._get_default_converter") + @patch("infra.local_converter._build_docling_converter") def test_forwards_all_options_to_build_converter(self, mock_build, mock_get_default): mock_conv = MagicMock() mock_result = MagicMock() @@ -312,25 +312,21 @@ class TestServiceForwardsPipelineOptions: @patch("services.analysis_service.document_repo") @patch("services.analysis_service.analysis_repo") - @patch("services.analysis_service._run_analysis") @pytest.mark.asyncio async def test_create_passes_pipeline_options_to_run( - self, mock_run, mock_analysis_repo, mock_doc_repo, mock_doc, + self, mock_analysis_repo, mock_doc_repo, mock_doc, ): mock_doc_repo.find_by_id = AsyncMock(return_value=mock_doc) mock_analysis_repo.insert = AsyncMock() - # Patch _run_analysis as a coroutine that we can inspect - mock_run.return_value = None - from services import analysis_service + mock_converter = AsyncMock() + from services.analysis_service import AnalysisService + svc = AnalysisService(converter=mock_converter) opts = {"do_ocr": False, "table_mode": "fast"} - # We need to patch asyncio.create_task to capture the coroutine args with patch("services.analysis_service.asyncio.create_task") as mock_task: - await analysis_service.create("d1", pipeline_options=opts) - - # create_task should have been called with _run_analysis(...) + await svc.create("d1", pipeline_options=opts) mock_task.assert_called_once() @patch("services.analysis_service.document_repo") @@ -342,32 +338,36 @@ class TestServiceForwardsPipelineOptions: mock_doc_repo.find_by_id = AsyncMock(return_value=mock_doc) mock_analysis_repo.insert = AsyncMock() - from services import analysis_service + mock_converter = AsyncMock() + from services.analysis_service import AnalysisService + svc = AnalysisService(converter=mock_converter) with patch("services.analysis_service.asyncio.create_task") as mock_task: - await analysis_service.create("d1") + await svc.create("d1") mock_task.assert_called_once() @patch("services.analysis_service.analysis_repo") @patch("services.analysis_service.document_repo") - @patch("services.analysis_service.convert_document") @pytest.mark.asyncio async def test_run_analysis_forwards_options_to_convert( - self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job, + self, mock_doc_repo, mock_analysis_repo, mock_job, ): - from domain.parsing import ConversionResult, PageDetail + from domain.value_objects import ConversionResult, PageDetail mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) mock_analysis_repo.update_status = AsyncMock() mock_doc_repo.update_page_count = AsyncMock() - mock_convert.return_value = ConversionResult( + + mock_converter = AsyncMock() + mock_converter.convert.return_value = ConversionResult( page_count=1, content_markdown="# Test", content_html="

Test

", pages=[PageDetail(page_number=1, width=612.0, height=792.0)], ) - from services.analysis_service import _run_analysis + from services.analysis_service import AnalysisService + svc = AnalysisService(converter=mock_converter) opts = { "do_ocr": False, @@ -381,10 +381,10 @@ class TestServiceForwardsPipelineOptions: "images_scale": 2.0, } - await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", opts) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", opts) - mock_convert.assert_called_once() - call_args = mock_convert.call_args + mock_converter.convert.assert_called_once() + call_args = mock_converter.convert.call_args assert call_args[0][0] == "/tmp/test.pdf" conv_opts = call_args[0][1] assert conv_opts.do_ocr is False @@ -395,47 +395,50 @@ class TestServiceForwardsPipelineOptions: @patch("services.analysis_service.analysis_repo") @patch("services.analysis_service.document_repo") - @patch("services.analysis_service.convert_document") @pytest.mark.asyncio async def test_run_analysis_uses_defaults_when_no_options( - self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job, + self, mock_doc_repo, mock_analysis_repo, mock_job, ): - from domain.parsing import ConversionResult, PageDetail + from domain.value_objects import ConversionResult, PageDetail mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) mock_analysis_repo.update_status = AsyncMock() mock_doc_repo.update_page_count = AsyncMock() - mock_convert.return_value = ConversionResult( + + mock_converter = AsyncMock() + mock_converter.convert.return_value = ConversionResult( page_count=1, content_markdown="", content_html="", pages=[PageDetail(page_number=1, width=612.0, height=792.0)], ) - from services.analysis_service import _run_analysis + from services.analysis_service import AnalysisService + svc = AnalysisService(converter=mock_converter) - await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", None) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", None) - # Called with file_path and default ConversionOptions - mock_convert.assert_called_once() - call_args = mock_convert.call_args + mock_converter.convert.assert_called_once() + call_args = mock_converter.convert.call_args assert call_args[0][0] == "/tmp/test.pdf" assert call_args[0][1] == ConversionOptions() @patch("services.analysis_service.analysis_repo") @patch("services.analysis_service.document_repo") - @patch("services.analysis_service.convert_document") @pytest.mark.asyncio async def test_run_analysis_marks_failed_on_error( - self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job, + self, mock_doc_repo, mock_analysis_repo, mock_job, ): mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) mock_analysis_repo.update_status = AsyncMock() - mock_convert.side_effect = RuntimeError("Docling crashed") - from services.analysis_service import _run_analysis + mock_converter = AsyncMock() + mock_converter.convert.side_effect = RuntimeError("Docling crashed") - await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"do_ocr": False}) + from services.analysis_service import AnalysisService + svc = AnalysisService(converter=mock_converter) + + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"do_ocr": False}) # Should have called update_status twice: RUNNING then FAILED assert mock_analysis_repo.update_status.call_count == 2 @@ -458,7 +461,7 @@ class TestAnalysisEndpointPipelineOptions: from main import app return TestClient(app, raise_server_exceptions=False) - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_no_pipeline_options_sends_none(self, mock_create, client): from domain.models import AnalysisJob mock_create.return_value = AnalysisJob(id="j1", document_id="d1") @@ -467,7 +470,7 @@ class TestAnalysisEndpointPipelineOptions: mock_create.assert_called_once_with("d1", pipeline_options=None) - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_empty_pipeline_options_object_uses_defaults(self, mock_create, client): from domain.models import AnalysisJob mock_create.return_value = AnalysisJob(id="j1", document_id="d1") @@ -485,7 +488,7 @@ class TestAnalysisEndpointPipelineOptions: assert opts["do_formula_enrichment"] is False assert opts["images_scale"] == 1.0 - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_partial_pipeline_options_merges_with_defaults(self, mock_create, client): from domain.models import AnalysisJob mock_create.return_value = AnalysisJob(id="j1", document_id="d1") @@ -508,7 +511,7 @@ class TestAnalysisEndpointPipelineOptions: assert opts["generate_picture_images"] is False assert opts["generate_page_images"] is False - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_full_pipeline_options(self, mock_create, client): from domain.models import AnalysisJob mock_create.return_value = AnalysisJob(id="j1", document_id="d1") @@ -535,7 +538,7 @@ class TestAnalysisEndpointPipelineOptions: opts = mock_create.call_args.kwargs["pipeline_options"] assert opts == payload["pipelineOptions"] - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_invalid_pipeline_option_type_rejected(self, mock_create, client): resp = client.post("/api/analyses", json={ "documentId": "d1", @@ -543,7 +546,7 @@ class TestAnalysisEndpointPipelineOptions: }) assert resp.status_code == 422 - @patch("services.analysis_service.create", new_callable=AsyncMock) + @patch("main.analysis_service.create", new_callable=AsyncMock) def test_unknown_pipeline_option_ignored(self, mock_create, client): from domain.models import AnalysisJob mock_create.return_value = AnalysisJob(id="j1", document_id="d1")