Refactor OcrmypdfPluginManager to use composition over inheritance

Replace inheritance from pluggy.PluginManager with composition pattern,
providing a type-safe interface for all 16 hooks defined in pluginspec.py.
The underlying pluggy manager is now accessible via the .pluggy property
for advanced use cases like set_blocked().

This change enables IDE autocomplete and type checking for all hook calls
while maintaining full backward compatibility with the plugin system.
This commit is contained in:
James R. Barlow
2026-01-07 17:23:13 -08:00
parent 0e946a7498
commit f5617ce44e
9 changed files with 232 additions and 66 deletions
+2 -2
View File
@@ -101,8 +101,8 @@ class Executor(ABC):
def setup_executor(plugin_manager) -> Executor:
pbar_class = plugin_manager.hook.get_progressbar_class()
return plugin_manager.hook.get_executor(progressbar_class=pbar_class)
pbar_class = plugin_manager.get_progressbar_class()
return plugin_manager.get_executor(progressbar_class=pbar_class)
class SerialExecutor(Executor):
+2 -2
View File
@@ -47,7 +47,7 @@ def get_docinfo(base_pdf: Pdf, context: PdfContext) -> dict[str, str]:
if options.subject:
pdfmark['/Subject'] = options.subject
creator_tag = context.plugin_manager.hook.get_ocr_engine().creator_tag(options)
creator_tag = context.plugin_manager.get_ocr_engine().creator_tag(options)
pdfmark['/Creator'] = f'{PROGRAM_NAME} {OCRMYPF_VERSION} / {creator_tag}'
pdfmark['/Producer'] = f'pikepdf {PIKEPDF_VERSION}'
@@ -186,7 +186,7 @@ def metadata_fixup(
output_file = context.get_path('metafix.pdf')
options = context.options
pbar_class = context.plugin_manager.hook.get_progressbar_class()
pbar_class = context.plugin_manager.get_progressbar_class()
with (
Pdf.open(context.origin) as original,
Pdf.open(working_file) as pdf,
+12 -12
View File
@@ -253,7 +253,7 @@ def validate_pdfinfo_options(context: PdfContext) -> None:
)
else:
raise TaggedPDFError()
context.plugin_manager.hook.validate(pdfinfo=pdfinfo, options=options)
context.plugin_manager.validate(pdfinfo=pdfinfo, options=options)
def _vector_page_dpi(pageinfo: PageInfo) -> int:
@@ -393,7 +393,7 @@ def rasterize_preview(input_file: Path, page_context: PageContext) -> Path:
[get_canvas_square_dpi(page_context)]
)
page_dpi = Resolution(300.0, 300.0).take_min([get_page_square_dpi(page_context)])
page_context.plugin_manager.hook.rasterize_pdf_page(
page_context.plugin_manager.rasterize_pdf_page(
input_file=input_file,
output_file=output_file,
raster_device='jpeggray',
@@ -453,7 +453,7 @@ def get_orientation_correction(preview: Path, page_context: PageContext) -> int:
which points it (hopefully) upright. _graft.py takes care of the orienting
the image and text layers.
"""
orient_conf = page_context.plugin_manager.hook.get_ocr_engine().get_orientation(
orient_conf = page_context.plugin_manager.get_ocr_engine().get_orientation(
preview, page_context.options
)
@@ -556,7 +556,7 @@ def rasterize(
canvas_dpi, page_dpi = calculate_raster_dpi(page_context)
page_context.plugin_manager.hook.rasterize_pdf_page(
page_context.plugin_manager.rasterize_pdf_page(
input_file=input_file,
output_file=output_file,
raster_device=device,
@@ -596,7 +596,7 @@ def preprocess_deskew(input_file: Path, page_context: PageContext) -> Path:
output_file = page_context.get_path('pp_deskew.png')
dpi = get_page_square_dpi(page_context, calculate_image_dpi(page_context))
ocr_engine = page_context.plugin_manager.hook.get_ocr_engine()
ocr_engine = page_context.plugin_manager.get_ocr_engine()
deskew_angle_degrees = ocr_engine.get_deskew(input_file, page_context.options)
with Image.open(input_file) as im:
@@ -661,7 +661,7 @@ def create_ocr_image(image: Path, page_context: PageContext) -> Path:
draw.rectangle(pixcoords, fill='white')
# draw.rectangle(pixcoords, outline='pink')
filter_im = page_context.plugin_manager.hook.filter_ocr_image(
filter_im = page_context.plugin_manager.filter_ocr_image(
page=page_context, image=im
)
if filter_im is not None:
@@ -679,7 +679,7 @@ def ocr_engine_hocr(input_file: Path, page_context: PageContext) -> tuple[Path,
hocr_text_out = page_context.get_path('ocr_hocr.txt')
options = page_context.options
ocr_engine = page_context.plugin_manager.hook.get_ocr_engine()
ocr_engine = page_context.plugin_manager.get_ocr_engine()
ocr_engine.generate_hocr(
input_file=input_file,
output_hocr=hocr_out,
@@ -766,7 +766,7 @@ def create_pdf_page_from_image(
bio.seek(0)
fix_pagepdf_boxes(bio, output_file, page_context, swap_axis=swap_axis)
output_file = page_context.plugin_manager.hook.filter_pdf_page(
output_file = page_context.plugin_manager.filter_pdf_page(
page=page_context, image_filename=image, output_pdf=output_file
)
return output_file
@@ -780,7 +780,7 @@ def ocr_engine_textonly_pdf(
output_text = page_context.get_path('ocr_tess.txt')
options = page_context.options
ocr_engine = page_context.plugin_manager.hook.get_ocr_engine()
ocr_engine = page_context.plugin_manager.get_ocr_engine()
ocr_engine.generate_pdf(
input_file=input_image,
output_pdf=output_pdf,
@@ -914,7 +914,7 @@ def convert_to_pdfa(input_pdf: Path, input_ps_stub: Path, context: PdfContext) -
else:
pdfa_part = '2' # Fallback
context.plugin_manager.hook.generate_pdfa(
context.plugin_manager.generate_pdfa(
pdf_version=input_pdfinfo.min_version,
pdf_pages=[fix_docinfo_file],
pdfmark=input_ps_stub,
@@ -922,7 +922,7 @@ def convert_to_pdfa(input_pdf: Path, input_ps_stub: Path, context: PdfContext) -
context=context,
pdfa_part=pdfa_part,
progressbar_class=(
context.plugin_manager.hook.get_progressbar_class()
context.plugin_manager.get_progressbar_class()
if options.progress_bar
else None
),
@@ -994,7 +994,7 @@ def optimize_pdf(
) -> tuple[Path, Sequence[str]]:
"""Optimize the given PDF file."""
output_file = context.get_path('optimize.pdf')
output_pdf, messages = context.plugin_manager.hook.optimize_pdf(
output_pdf, messages = context.plugin_manager.optimize_pdf(
input_pdf=input_file,
output_pdf=output_file,
context=context,
+2 -2
View File
@@ -445,7 +445,7 @@ def process_page(page_context: PageContext) -> tuple[Path, Path | None, int]:
visible_image_out = preprocess_out
if should_visible_page_image_use_jpg(page_context.pageinfo):
visible_image_out = create_visible_page_jpg(visible_image_out, page_context)
filtered_image = page_context.plugin_manager.hook.filter_page_image(
filtered_image = page_context.plugin_manager.filter_page_image(
page=page_context, image_filename=visible_image_out
)
if filtered_image is not None: # None if no hook is present
@@ -472,7 +472,7 @@ def postprocess(
ps_stub_out = generate_postscript_stub(context)
pdf_out = convert_to_pdfa(pdf_out, ps_stub_out, context)
optimizing = context.plugin_manager.hook.is_optimization_enabled(context=context)
optimizing = context.plugin_manager.is_optimization_enabled(context=context)
save_settings = get_pdf_save_settings(context.options.output_type)
save_settings['linearize'] = not optimizing and should_linearize(pdf_out, context)
+1 -1
View File
@@ -117,7 +117,7 @@ def run_hocr_to_ocr_pdf_pipeline(
# Gather pdfinfo and create context
pdfinfo = do_get_pdfinfo(origin_pdf, executor, options)
context = PdfContext(options, work_folder, origin_pdf, pdfinfo, plugin_manager)
plugin_manager.hook.check_options(options=options)
plugin_manager.check_options(options=options)
optimize_messages = exec_hocr_to_ocr_pdf(context, executor)
return report_output_pdf(options, origin_pdf, optimize_messages)
+192 -25
View File
@@ -1,7 +1,7 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Plugin manager using pluggy."""
"""Plugin manager using pluggy with type-safe interface."""
from __future__ import annotations
@@ -9,23 +9,36 @@ import importlib
import importlib.util
import pkgutil
import sys
from argparse import ArgumentParser
from collections.abc import Sequence
from logging import Handler
from pathlib import Path
from typing import TYPE_CHECKING
import pluggy
from pydantic import BaseModel
import ocrmypdf.builtin_plugins
from ocrmypdf import pluginspec
from ocrmypdf import Executor, PdfContext, pluginspec
from ocrmypdf._options import OCROptions
from ocrmypdf._progressbar import ProgressBar
from ocrmypdf.helpers import Resolution
from ocrmypdf.pluginspec import OcrEngine
if TYPE_CHECKING:
from PIL import Image
from ocrmypdf._jobcontext import PageContext
from ocrmypdf.pdfinfo import PdfInfo
class OcrmypdfPluginManager(pluggy.PluginManager):
"""pluggy.PluginManager that can fork.
class OcrmypdfPluginManager:
"""Type-safe wrapper around pluggy.PluginManager.
Capable of reconstructing itself in child workers.
Capable of reconstructing itself in child workers via pickle.
Arguments:
setup_func: callback that initializes the plugin manager with all
standard plugins
This class provides type-safe methods for all hooks defined in pluginspec.py,
removing the need for unsafe `hook.method_name()` calls.
"""
def __init__(
@@ -35,19 +48,28 @@ class OcrmypdfPluginManager(pluggy.PluginManager):
builtins: bool = True,
**kwargs,
):
self.__init_args = args
self.__init_kwargs = kwargs
self.__plugins = plugins
self.__builtins = builtins
super().__init__(*args, **kwargs)
self.setup_plugins()
self._init_args = args
self._init_kwargs = kwargs
self._plugins = plugins
self._builtins = builtins
self._pm = pluggy.PluginManager(*args, **kwargs)
self._setup_plugins()
@property
def pluggy(self) -> pluggy.PluginManager:
"""Access the underlying pluggy.PluginManager for advanced use cases.
This is useful for plugins that need to call methods like set_blocked()
in their initialize hook.
"""
return self._pm
def __getstate__(self):
state = dict(
init_args=self.__init_args,
plugins=self.__plugins,
builtins=self.__builtins,
init_kwargs=self.__init_kwargs,
init_args=self._init_args,
plugins=self._plugins,
builtins=self._builtins,
init_kwargs=self._init_kwargs,
)
return state
@@ -59,23 +81,23 @@ class OcrmypdfPluginManager(pluggy.PluginManager):
**state['init_kwargs'],
)
def setup_plugins(self):
self.add_hookspecs(pluginspec)
def _setup_plugins(self):
self._pm.add_hookspecs(pluginspec)
# 1. Register builtins
if self.__builtins:
if self._builtins:
for module in sorted(
pkgutil.iter_modules(ocrmypdf.builtin_plugins.__path__)
):
name = f'ocrmypdf.builtin_plugins.{module.name}'
module = importlib.import_module(name)
self.register(module)
self._pm.register(module)
# 2. Register setuptools plugins
self.load_setuptools_entrypoints('ocrmypdf')
self._pm.load_setuptools_entrypoints('ocrmypdf')
# 3. Register plugins specified on command line
for name in self.__plugins:
for name in self._plugins:
if isinstance(name, Path) or name.endswith('.py'):
# Import by filename
module_name = Path(name).stem
@@ -86,7 +108,152 @@ class OcrmypdfPluginManager(pluggy.PluginManager):
else:
# Import by dotted module name
module = importlib.import_module(name)
self.register(module)
self._pm.register(module)
# =========================================================================
# Type-safe hook methods
# =========================================================================
# --- firstresult hooks ---
def get_logging_console(self) -> Handler | None:
"""Returns a custom logging handler for progress bar compatibility."""
return self._pm.hook.get_logging_console()
def get_executor(self, *, progressbar_class: type[ProgressBar]) -> Executor | None:
"""Returns an executor for parallel processing."""
return self._pm.hook.get_executor(progressbar_class=progressbar_class)
def get_progressbar_class(self) -> type[ProgressBar] | None:
"""Returns a progress bar class."""
return self._pm.hook.get_progressbar_class()
def rasterize_pdf_page(
self,
*,
input_file: Path,
output_file: Path,
raster_device: str,
raster_dpi: Resolution,
pageno: int,
page_dpi: Resolution | None,
rotation: int | None,
filter_vector: bool,
stop_on_soft_error: bool,
options: OCROptions | None,
use_cropbox: bool,
) -> Path | None:
"""Rasterize one page of a PDF at specified resolution."""
return self._pm.hook.rasterize_pdf_page(
input_file=input_file,
output_file=output_file,
raster_device=raster_device,
raster_dpi=raster_dpi,
pageno=pageno,
page_dpi=page_dpi,
rotation=rotation,
filter_vector=filter_vector,
stop_on_soft_error=stop_on_soft_error,
options=options,
use_cropbox=use_cropbox,
)
def filter_ocr_image(
self, *, page: PageContext, image: Image.Image
) -> Image.Image | None:
"""Filter the image before it is sent to OCR."""
return self._pm.hook.filter_ocr_image(page=page, image=image)
def filter_page_image(
self, *, page: PageContext, image_filename: Path
) -> Path | None:
"""Filter the whole page image before it is inserted into the PDF."""
return self._pm.hook.filter_page_image(page=page, image_filename=image_filename)
def filter_pdf_page(
self, *, page: PageContext, image_filename: Path, output_pdf: Path
) -> Path | None:
"""Convert a filtered whole page image into a PDF."""
return self._pm.hook.filter_pdf_page(
page=page, image_filename=image_filename, output_pdf=output_pdf
)
def get_ocr_engine(self) -> OcrEngine | None:
"""Returns an OcrEngine to use for processing."""
return self._pm.hook.get_ocr_engine()
def generate_pdfa(
self,
*,
pdf_pages: list[Path],
pdfmark: Path,
output_file: Path,
context: PdfContext,
pdf_version: str,
pdfa_part: str,
progressbar_class: type[ProgressBar] | None,
stop_on_soft_error: bool,
) -> Path | None:
"""Generate a PDF/A file."""
return self._pm.hook.generate_pdfa(
pdf_pages=pdf_pages,
pdfmark=pdfmark,
output_file=output_file,
context=context,
pdf_version=pdf_version,
pdfa_part=pdfa_part,
progressbar_class=progressbar_class,
stop_on_soft_error=stop_on_soft_error,
)
def optimize_pdf(
self,
*,
input_pdf: Path,
output_pdf: Path,
context: PdfContext,
executor: Executor,
linearize: bool,
) -> tuple[Path, Sequence[str]] | None:
"""Optimize a PDF after OCR processing."""
return self._pm.hook.optimize_pdf(
input_pdf=input_pdf,
output_pdf=output_pdf,
context=context,
executor=executor,
linearize=linearize,
)
def is_optimization_enabled(self, *, context: PdfContext) -> bool | None:
"""Returns whether optimization is enabled for given context."""
return self._pm.hook.is_optimization_enabled(context=context)
# --- non-firstresult hooks ---
def initialize(self, *, plugin_manager: pluggy.PluginManager) -> list[None]:
"""Called when plugins are first loaded.
Args:
plugin_manager: The underlying pluggy.PluginManager, allowing
plugins to call methods like set_blocked().
"""
return self._pm.hook.initialize(plugin_manager=plugin_manager)
def add_options(self, *, parser: ArgumentParser) -> list[None]:
"""Allows plugins to add command line and API arguments."""
return self._pm.hook.add_options(parser=parser)
def register_options(self) -> list[dict[str, type[BaseModel]]]:
"""Returns plugin option models keyed by namespace."""
return self._pm.hook.register_options()
def check_options(self, *, options: OCROptions) -> list[None]:
"""Called to validate options after parsing."""
return self._pm.hook.check_options(options=options)
def validate(self, *, pdfinfo: PdfInfo, options: OCROptions) -> list[None]:
"""Called to validate options and pdfinfo after PDF is loaded."""
return self._pm.hook.validate(pdfinfo=pdfinfo, options=options)
def get_plugin_manager(
+7 -5
View File
@@ -14,9 +14,9 @@ from pathlib import Path
from shutil import copyfileobj
import pikepdf
from pluggy import PluginManager
from ocrmypdf._defaults import DEFAULT_ROTATE_PAGES_THRESHOLD
from ocrmypdf._plugin_manager import OcrmypdfPluginManager
from ocrmypdf._exec import unpaper
from ocrmypdf._options import OCROptions
from ocrmypdf.exceptions import (
@@ -120,12 +120,14 @@ def _check_plugin_invariant_options(options: OCROptions) -> None:
check_options_preprocessing(options)
def _check_plugin_options(options: OCROptions, plugin_manager: PluginManager) -> None:
def _check_plugin_options(
options: OCROptions, plugin_manager: OcrmypdfPluginManager
) -> None:
# First, let plugins check their external dependencies
plugin_manager.hook.check_options(options=options)
plugin_manager.check_options(options=options)
# Then check OCR engine language support
ocr_engine_languages = plugin_manager.hook.get_ocr_engine().languages(options)
ocr_engine_languages = plugin_manager.get_ocr_engine().languages(options)
check_options_languages(options, ocr_engine_languages)
# Finally, run comprehensive validation using the coordinator
@@ -134,7 +136,7 @@ def _check_plugin_options(options: OCROptions, plugin_manager: PluginManager) ->
coordinator.validate_all_options(options)
def check_options(options: OCROptions, plugin_manager: PluginManager) -> None:
def check_options(options: OCROptions, plugin_manager: OcrmypdfPluginManager) -> None:
"""Check options for validity and consistency.
This function coordinates validation across the entire system:
+11 -13
View File
@@ -50,14 +50,12 @@ from pathlib import Path
from typing import BinaryIO
from warnings import warn
import pluggy
from ocrmypdf._logging import PageNumberFilter
from ocrmypdf._options import OCROptions
from ocrmypdf._pipelines.hocr_to_ocr_pdf import run_hocr_to_ocr_pdf_pipeline
from ocrmypdf._pipelines.ocr import run_pipeline, run_pipeline_cli
from ocrmypdf._pipelines.pdf_to_hocr import run_hocr_pipeline
from ocrmypdf._plugin_manager import get_plugin_manager
from ocrmypdf._plugin_manager import OcrmypdfPluginManager, get_plugin_manager
from ocrmypdf._validation import check_options
from ocrmypdf.cli import ArgumentParser, get_parser
@@ -72,8 +70,8 @@ _api_lock = threading.Lock()
def setup_plugin_infrastructure(
plugins: Sequence[Path | str] | None = None,
plugin_manager: pluggy.PluginManager | None = None,
) -> pluggy.PluginManager:
plugin_manager: OcrmypdfPluginManager | None = None,
) -> OcrmypdfPluginManager:
"""Set up plugin infrastructure with proper initialization.
This function handles:
@@ -105,8 +103,8 @@ def setup_plugin_infrastructure(
if not plugin_manager:
plugin_manager = get_plugin_manager(plugins)
# Initialize plugins
plugin_manager.hook.initialize(plugin_manager=plugin_manager) # pylint: disable=no-member
# Initialize plugins (pass the underlying pluggy manager)
plugin_manager.initialize(plugin_manager=plugin_manager.pluggy)
# Initialize plugin option registry
from ocrmypdf._plugin_registry import PluginOptionRegistry
@@ -114,7 +112,7 @@ def setup_plugin_infrastructure(
registry = PluginOptionRegistry()
# Let plugins register their option models
option_models = plugin_manager.hook.register_options() # pylint: disable=no-member
option_models = plugin_manager.register_options()
all_plugin_models: dict[str, type] = {}
for plugin_options in option_models:
if plugin_options: # Skip None returns
@@ -146,7 +144,7 @@ def configure_logging(
*,
progress_bar_friendly: bool = True,
manage_root_logger: bool = False,
plugin_manager: pluggy.PluginManager | None = None,
plugin_manager: OcrmypdfPluginManager | None = None,
):
"""Set up logging.
@@ -193,7 +191,7 @@ def configure_logging(
console = None
if plugin_manager and progress_bar_friendly:
console = plugin_manager.hook.get_logging_console()
console = plugin_manager.get_logging_console()
if not console:
console = logging.StreamHandler(stream=sys.stderr)
@@ -432,7 +430,7 @@ def ocr( # noqa: D417
# Get parser and let plugins add their options
parser = get_parser()
plugin_manager.hook.add_options(parser=parser) # pylint: disable=no-member
plugin_manager.add_options(parser=parser)
if 'verbose' in kwargs:
warn("ocrmypdf.ocr(verbose=) is ignored. Use ocrmypdf.configure_logging().")
@@ -574,7 +572,7 @@ def _pdf_to_hocr( # noqa: D417
plugins=plugins, plugin_manager=plugin_manager
)
plugin_manager.hook.add_options(parser=get_parser()) # pylint: disable=no-member
plugin_manager.add_options(parser=get_parser())
# Create OCROptions directly
try:
@@ -687,7 +685,7 @@ def _hocr_to_ocr_pdf( # noqa: D417
plugins=plugins, plugin_manager=plugin_manager
)
plugin_manager.hook.add_options(parser=get_parser()) # pylint: disable=no-member
plugin_manager.add_options(parser=get_parser())
# Create OCROptions directly
try:
+3 -4
View File
@@ -10,11 +10,10 @@ from argparse import ArgumentParser
from collections.abc import Callable, Mapping
from typing import Any, TypeVar
import pluggy
from ocrmypdf._defaults import DEFAULT_ROTATE_PAGES_THRESHOLD
from ocrmypdf._defaults import PROGRAM_NAME as _PROGRAM_NAME
from ocrmypdf._options import OCROptions
from ocrmypdf._plugin_manager import OcrmypdfPluginManager
from ocrmypdf._version import __version__ as _VERSION
T = TypeVar('T', int, float)
@@ -480,7 +479,7 @@ def namespace_to_options(ns) -> OCROptions:
def get_options_and_plugins(
args=None,
) -> tuple[OCROptions, pluggy.PluginManager]:
) -> tuple[OCROptions, OcrmypdfPluginManager]:
"""Parse command line arguments and return OCROptions and plugin manager.
This is the main entry point for CLI argument processing. It handles
@@ -504,7 +503,7 @@ def get_options_and_plugins(
# Get parser and let plugins add their options
parser = get_parser()
plugin_manager.hook.add_options(parser=parser) # pylint: disable=no-member
plugin_manager.add_options(parser=parser)
# Parse all arguments
namespace = parser.parse_args(args=args)