fix(passport-worker): move PDF conversion config on the worker config side

Author: ClemDoum

workers/passport-worker/passport_worker/activities.py

@@ -38,7 +38,6 @@
     ImagePreprocessorConfig,
     PassportDetectionArgs,
     PassportDetectionResponse,
-    PDFConverterConfig,
     PreprocessingBatches,
 )
 from .preprocessing import (
@@ -147,30 +146,29 @@ async def convert_to_pdfs(
         self,
         batch: Path,
         project: str,
-        config: PDFConverterConfig,
         *,
         progress: Annotated[
             AsyncProgressRateHandler | None, Weight(value=_CONVERT_TO_PDF_WEIGHT)
         ] = None,
     ) -> tuple[Path, Path]:
         worker_config = cast(PassportWorkerConfig, lifespan_worker_config())
+        config = worker_config.preprocessing.pdfs
         cache = lifespan_pdf_converter_cache()
-        pdf_converter_cache_key = config_cache_key(config)
+        pdf_converter_cache_key = config_cache_key(config.pdf_converter)
         pdf_converter_factory = async_enter_cm(
-            partial(PDFConverter.from_config, config)
+            partial(PDFConverter.from_config, config.pdf_converter)
         )
         pdf_converter = await cache.async_get_or_cache_resource(
             pdf_converter_cache_key, pdf_converter_factory
         )
         workdir = worker_config.paths.workdir
         pdfs_root = activity_workdir(workdir, project, act_context=False)
         pdfs_root.mkdir(parents=True, exist_ok=True)
-        max_concurrency = worker_config.preprocessing.pdfs.max_concurrency
         successes, errors = await convert_to_pdfs_act(
             batch,
             pdf_converter,
             worker_config.paths,
-            max_concurrency,
+            config.max_concurrency,
             output_root=pdfs_root,
             progress=progress,
         )

workers/passport-worker/passport_worker/config.py

@@ -1,4 +1,6 @@
 from concurrent.futures import ProcessPoolExecutor
+from enum import StrEnum
+from typing import ClassVar
 
 import datashare_python
 from datashare_python.config import (
@@ -8,6 +10,7 @@
     WorkerConfig,
 )
 from datashare_python.objects import DatashareModel, WorkerPaths
+from icij_common.registrable import RegistrableConfig
 from pydantic import Field
 
 _ALL_LOGGERS = [datashare_python.__name__, __name__, "__main__"]
@@ -31,9 +34,37 @@ def to_image_preprocessing_executor(self) -> ProcessPoolExecutor:
         return ProcessPoolExecutor(max_workers=self.n_processes)
 
 
+class PDFConverterType(StrEnum):
+    GOTENBERG = "gotenberg"
+
+
+class PDFConverterConfigBase(DatashareModel, RegistrableConfig):
+    registry_key: ClassVar[str] = Field(frozen=True, default="type")
+    type: ClassVar[PDFConverterType]
+
+
+class GotenbergPDFConverterConfig(PDFConverterConfigBase):
+    type: ClassVar[PDFConverterType] = Field(
+        frozen=True, default=PDFConverterType.GOTENBERG
+    )
+    gotenberg_url: str = "http://localhost:3000"
+    max_retries: int = 5
+    min_retry_wait_s: float = 5.0
+    max_retry_wait_s: float = 30.0
+    max_retry_randomness_s: float = 2.0
+
+
+# TODO: use a tagged union here when we have more options
+PDFConverterConfig = GotenbergPDFConverterConfig
+
+
 class PDFConversionWorkerConfig(DatashareModel):
     max_concurrency: int = 10
 
+    pdf_converter: PDFConverterConfig = Field(
+        default_factory=GotenbergPDFConverterConfig
+    )
+
 
 class PreprocessingWorkerConfig(DatashareModel):
     target_n_pages_per_batch: int = 200

workers/passport-worker/passport_worker/objects.py

@@ -60,36 +60,10 @@ class DefaultImagePreprocessorConfig(ImagePreprocessorConfigBase):
 ImagePreprocessorConfig = DefaultImagePreprocessorConfig
 
 
-class PDFConverterType(StrEnum):
-    GOTENBERG = "gotenberg"
-
-
-class PDFConverterConfigBase(DatashareModel, RegistrableConfig):
-    registry_key: ClassVar[str] = Field(frozen=True, default="type")
-    type: ClassVar[PDFConverterType]
-
-
-class GotenbergPDFConverterConfig(PDFConverterConfigBase):
-    type: ClassVar[PDFConverterType] = Field(
-        frozen=True, default=PDFConverterType.GOTENBERG
-    )
-
-    gotenberg_url: str = "http://localhost:3000"
-    max_retries: int = 5
-    min_retry_wait_s: float = 5.0
-    max_retry_wait_s: float = 30.0
-    max_retry_randomness_s: float = 2.0
-
-
-# TODO: use a tagged union here when we have more options
-PDFConverterConfig = GotenbergPDFConverterConfig
-
-
 class PreprocessingConfig(DatashareModel):
     images: ImagePreprocessorConfig = Field(
         default_factory=DefaultImagePreprocessorConfig
     )
-    pdfs: PDFConverterConfig = Field(default_factory=GotenbergPDFConverterConfig)
 
 
 class PassportDetectorType(StrEnum):

workers/passport-worker/passport_worker/preprocessing.py

@@ -6,7 +6,7 @@
 from inspect import iscoroutinefunction
 from pathlib import Path
 from types import TracebackType
-from typing import ClassVar, Protocol, Self, TypeVar
+from typing import Protocol, Self, TypeVar
 
 from aiofile import async_open
 from datashare_python.objects import ProcessedPage, WorkerPaths
@@ -18,21 +18,19 @@
     to_raw_async_progress,
     to_raw_sync_progress,
 )
-from icij_common.registrable import RegistrableConfig, RegistrableFromConfig
+from icij_common.registrable import RegistrableFromConfig
 from passport_service import GotenbergClient
 from passport_service.constants import Colorspace
 from passport_service.core import process_image, process_pdf
 from passport_service.exceptions import UnsupportedDocExtension
 from passport_service.utils import run_with_concurrency
-from pydantic import Field
 
+from passport_worker.config import GotenbergPDFConverterConfig, PDFConverterType
 from passport_worker.constants import pil_supported_extensions
 from passport_worker.objects import (
     DefaultImagePreprocessorConfig,
     FileProcessingError,
-    GotenbergPDFConverterConfig,
     ImagePreprocessorType,
-    PDFConverterType,
     ProcessedFile,
 )
 
@@ -76,11 +74,6 @@ def _from_config(cls, config: DefaultImagePreprocessorConfig, **extras) -> Self:
         return cls(config)
 
 
-class PDFConverterConfig(RegistrableConfig):
-    registry_key: ClassVar[str] = Field(frozen=True, default="type")
-    type: ClassVar[PDFConverterType]
-
-
 class PDFConverter(RegistrableFromConfig):
     max_concurrency: int = 10
 

workers/passport-worker/passport_worker/workflows.py

@@ -16,7 +16,6 @@
         ImagePreprocessorConfig,
         PassportDetectionArgs,
         PassportDetectionResponse,
-        PDFConverterConfig,
         PreprocessingBatches,
     )
 
@@ -99,7 +98,7 @@ async def preprocess(
         preprocessing_batches.images, args.project, args.config.preprocessing.images
     )
     convert_to_pdf_tasks = _convert_to_pdfs_tasks(
-        preprocessing_batches.to_pdf, args.project, args.config.preprocessing.pdfs
+        preprocessing_batches.to_pdf, args.project
     )
     im_preprocessing_tasks = asyncio.gather(*im_preprocessing_tasks)
     convert_to_pdf_tasks = asyncio.gather(*convert_to_pdf_tasks)
@@ -157,15 +156,13 @@ def _im_processing_tasks(
     return im_preprocessing_tasks
 
 
-def _convert_to_pdfs_tasks(
-    batches: Batches, project: str, config: PDFConverterConfig
-) -> list[Coroutine]:
+def _convert_to_pdfs_tasks(batches: Batches, project: str) -> list[Coroutine]:
     all_tasks = []
     for b in batches:
         all_tasks.append(
             execute_activity(
                 PassportDetectionActivities.convert_to_pdfs,
-                args=(b, project, config),
+                args=(b, project),
                 task_queue=TaskQueue.IO,
                 start_to_close_timeout=_CONVERT_TO_PDF_TIMEOUT,
             )