Parse command middleware loading vs config

Author: avvertix

src/parxy_cli/commands/parse.py

@@ -1,11 +1,10 @@
 """Command line interface for Parxy document processing."""
 
 import json
-import os
 import tomllib
 from datetime import timedelta
 from pathlib import Path
-from typing import Optional, List, Annotated, Dict
+from typing import Optional, List, Annotated
 
 import typer
 
@@ -20,31 +19,21 @@
 console = Console()
 
 
-DEFAULT_MIDDLEWARE_PROFILES: Dict[str, List[str]] = {
-    'default': [],
-}
-
-
-def _load_middleware_profiles(config_path: Optional[Path]) -> Dict[str, List[str]]:
-    """Load middleware profiles from optional config file.
+def _load_middleware_from_config(config_path: Path) -> List[str]:
+    """Load middleware class paths from a config file.
 
     Supports JSON, TOML, YAML and YML. The expected structure is either:
-    - ``{"middleware_profiles": {"profile": ["path.to.Middleware"]}}``
-    - ``{"profiles": {"profile": ["path.to.Middleware"]}}``
+    - A top-level list: ``["path.to.Middleware1", "path.to.Middleware2"]``
+    - An object with a ``middleware`` key: ``{"middleware": ["path.to.Middleware"]}``
     """
-    profiles = dict(DEFAULT_MIDDLEWARE_PROFILES)
-    if config_path is None:
-        return profiles
-
     if not config_path.exists():
         raise typer.BadParameter(f'Middleware config file not found: {config_path}')
 
     suffix = config_path.suffix.lower()
-    raw_data = None
 
-    if suffix in {'.json'}:
+    if suffix == '.json':
         raw_data = json.loads(config_path.read_text(encoding='utf-8'))
-    elif suffix in {'.toml'}:
+    elif suffix == '.toml':
         raw_data = tomllib.loads(config_path.read_text(encoding='utf-8'))
     elif suffix in {'.yaml', '.yml'}:
         try:
@@ -59,60 +48,43 @@ def _load_middleware_profiles(config_path: Optional[Path]) -> Dict[str, List[str
             'Unsupported middleware config format. Use .json, .toml, .yaml or .yml'
         )
 
-    if not isinstance(raw_data, dict):
-        raise typer.BadParameter(
-            'Middleware config must be an object/map at top level.'
-        )
-
-    custom_profiles = raw_data.get('middleware_profiles') or raw_data.get('profiles')
-    if custom_profiles is None:
-        return profiles
-
-    if not isinstance(custom_profiles, dict):
+    if isinstance(raw_data, list):
+        middleware_list = raw_data
+    elif isinstance(raw_data, dict):
+        middleware_list = raw_data.get('middleware', [])
+        if not isinstance(middleware_list, list):
+            raise typer.BadParameter(
+                'middleware_config: "middleware" key must be a list of class paths.'
+            )
+    else:
         raise typer.BadParameter(
-            'middleware_profiles must be a map of profile names to middleware paths.'
+            'Middleware config must be a list or an object with a "middleware" key.'
         )
 
-    for name, middleware_list in custom_profiles.items():
-        if not isinstance(name, str) or not isinstance(middleware_list, list):
-            raise typer.BadParameter(
-                'Each profile must map to a list of middleware class paths.'
-            )
-        if not all(isinstance(item, str) for item in middleware_list):
-            raise typer.BadParameter('Middleware class paths must be strings.')
-        profiles[name] = middleware_list
+    if not all(isinstance(item, str) for item in middleware_list):
+        raise typer.BadParameter('Middleware class paths must be strings.')
 
-    return profiles
+    return middleware_list
 
 
-def configure_middleware_profile(
-    profile: Optional[str],
+def configure_middleware(
+    middleware: Optional[List[str]],
     config_path: Optional[Path],
 ) -> None:
-    """Configure global middleware registry from a selected profile."""
-    selected_profile = profile or os.getenv('PARXY_MIDDLEWARE_PROFILE')
-    if not selected_profile:
-        return
+    """Configure global middleware from inline class paths and/or a config file."""
+    paths: List[str] = list(middleware or [])
 
-    env_config_path = os.getenv('PARXY_MIDDLEWARE_CONFIG')
-    effective_config = config_path or (
-        Path(env_config_path) if env_config_path else None
-    )
-    profiles = _load_middleware_profiles(effective_config)
+    if config_path is not None:
+        paths.extend(_load_middleware_from_config(config_path))
 
-    if selected_profile not in profiles:
-        available = ', '.join(sorted(profiles.keys()))
-        raise typer.BadParameter(
-            f'Unknown middleware profile: {selected_profile}. Available profiles: {available}'
-        )
+    if not paths:
+        return
 
-    middleware_paths = profiles[selected_profile]
     Parxy.clear_middleware()
-    if middleware_paths:
-        Parxy.with_middleware(middleware_paths)
+    Parxy.with_middleware(paths)
 
     console.info(
-        f"Using middleware profile '{selected_profile}' ({len(middleware_paths)} middleware)."
+        f'Using {len(paths)} middleware class{"es" if len(paths) != 1 else ""}.'
     )
 
 
@@ -360,20 +332,20 @@ def parse(
             min=1,
         ),
     ] = None,
-    middleware_profile: Annotated[
-        Optional[str],
+    middleware: Annotated[
+        Optional[List[str]],
         typer.Option(
-            '--middleware-profile',
-            envvar='PARXY_MIDDLEWARE_PROFILE',
-            help='Middleware profile name to activate. Built-in: default. Additional profiles can be loaded via --middleware-config.',
+            '--middleware',
+            '-p',
+            help='Middleware class path(s) to apply. Can be specified multiple times (e.g. --middleware my.pkg.MyMiddleware).',
         ),
     ] = None,
     middleware_config: Annotated[
         Optional[str],
         typer.Option(
             '--middleware-config',
             envvar='PARXY_MIDDLEWARE_CONFIG',
-            help='Path to a .json/.toml/.yaml config file defining custom middleware profiles.',
+            help='Path to a .json/.toml/.yaml file with a list of middleware class paths to apply. Appended after inline middleware with --middleware',
         ),
     ] = None,
 ):
@@ -427,9 +399,8 @@ def parse(
     # Calculate total tasks
     total_tasks = len(files) * len(drivers)
 
-
-    configure_middleware_profile(
-        profile=middleware_profile,
+    configure_middleware(
+        middleware=middleware,
         config_path=Path(middleware_config) if middleware_config else None,
     )
 

src/parxy_core/drivers/factory.py

@@ -48,8 +48,11 @@ class DriverFactory:
     __custom_creators: Dict[str, Callable[[], Driver]] = {}
     """The custom drivers"""
 
+    __config_middleware: List[Middleware] = []
+    """Middleware loaded from ParxyConfig — preserved across clear_middleware() calls."""
+
     __middleware: List[Middleware] = []
-    """The global middleware registry"""
+    """Runtime middleware added programmatically via with_middleware()."""
 
     _config: Optional[ParxyConfig] = None
 
@@ -79,6 +82,7 @@ def reset(cls):
         Useful for testing and isolation between test cases.
         """
         cls.__instance = None
+        cls.__config_middleware = []
         cls.__middleware = []
         cls.__drivers = {}
         cls.__custom_creators = {}
@@ -108,18 +112,18 @@ def initialize(self, config: ParxyConfig) -> Self:
         return self
 
     def _load_middleware_from_config(self) -> None:
-        """Load middleware from ParxyConfig.middleware.
+        """Load middleware from ParxyConfig.middleware into the config layer.
 
-        Middleware specified in config are automatically registered
-        in the factory's global middleware registry.
+        Config middleware is kept separate from runtime middleware so it
+        survives clear_middleware() calls.
         """
         if not self._config.middleware:
             return
 
         for middleware_path in self._config.middleware:
             try:
                 middleware = self._import_middleware(middleware_path)
-                self.__middleware.append(middleware)
+                self.__config_middleware.append(middleware)
                 self._logger.info(f'Loaded middleware from config: {middleware_path}')
             except (ImportError, ValueError) as e:
                 self._logger.warning(
@@ -342,7 +346,10 @@ def with_middleware(
         return self
 
     def clear_middleware(self) -> 'DriverFactory':
-        """Clear all global middleware.
+        """Clear runtime middleware.
+
+        Middleware loaded from ``ParxyConfig.middleware`` is preserved.
+        Only middleware added via :meth:`with_middleware` is removed.
 
         Returns
         -------
@@ -354,14 +361,14 @@ def clear_middleware(self) -> 'DriverFactory':
         return self
 
     def get_middleware(self) -> List[Middleware]:
-        """Get the list of global middleware.
+        """Get the combined middleware list (config layer + runtime layer).
 
         Returns
         -------
         List[Middleware]
-            Copy of the current global middleware list
+            Copy of the current middleware list, config entries first.
         """
-        return list(self.__middleware)
+        return list(self.__config_middleware) + list(self.__middleware)
 
     def get_drivers(self) -> Dict[str, Driver]:
         """Get all of the created "drivers".

src/parxy_core/facade/parxy.py

@@ -119,7 +119,7 @@ def with_middleware(cls, middleware: List[Union[str, Middleware]]) -> 'Parxy':
         Example
         -------
         >>> Parxy.with_middleware(
-        ...     [LoggingMiddleware(), 'parxy_core.middleware.PIIScanner']
+        ...     [MyCustomMiddleware(), 'parxy_core.middleware.SimpleMiddleware']
         ... )
         >>> doc = Parxy.parse('document.pdf')
         """

src/parxy_core/models/config.py

@@ -1,10 +1,11 @@
-from typing import Literal, Optional, List
+from typing import Literal, Optional, List, Any
 
+import json
 import logging
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from pydantic import Field, SecretStr, BaseModel
+from pydantic import Field, SecretStr, BaseModel, field_validator
 
 
 class BaseConfig(BaseSettings):
@@ -82,6 +83,16 @@ class ParxyConfig(BaseConfig):
     middleware: Optional[List[str]] = None
     """List of middleware class paths to load automatically."""
 
+    @field_validator('middleware', mode='before')
+    @classmethod
+    def parse_middleware(cls, v: Any) -> Any:
+        if not isinstance(v, str):
+            return v
+        stripped = v.strip()
+        if stripped.startswith('['):
+            return json.loads(stripped)
+        return [item.strip() for item in stripped.split(',') if item.strip()]
+
     model_config = SettingsConfigDict(
         env_prefix='parxy_',
         env_file='.env',

tests/commands/test_parse.py

@@ -239,45 +239,89 @@ def test_parse_command_with_multiple_drivers(runner, mock_document, tmp_path):
         assert (output_dir / 'llamaparse-test.json').exists()
 
 
-def test_parse_command_with_default_middleware_profile(runner, mock_document, tmp_path):
-    """Test middleware profile selection with builtin default profile."""
+def test_parse_command_with_middleware(runner, mock_document, tmp_path):
+    """Test inline middleware class paths via --middleware."""
 
     test_file = tmp_path / 'test.pdf'
     test_file.write_text('dummy pdf content')
 
     with patch('parxy_cli.commands.parse.Parxy') as mock_parxy:
-        mock_parxy.parse.return_value = mock_document
         mock_parxy.default_driver.return_value = 'pymupdf'
+        mock_parxy.batch_iter.return_value = iter(
+            [
+                BatchResult(
+                    file=str(test_file),
+                    driver='pymupdf',
+                    document=mock_document,
+                    error=None,
+                )
+            ]
+        )
+
+        result = runner.invoke(
+            app,
+            [str(test_file), '--middleware', 'parxy_core.middleware.SimpleMiddleware'],
+        )
+
+        assert result.exit_code == 0
+        mock_parxy.clear_middleware.assert_called_once()
+        mock_parxy.with_middleware.assert_called_once_with(
+            ['parxy_core.middleware.SimpleMiddleware']
+        )
+
+
+def test_parse_command_with_middleware_config(runner, mock_document, tmp_path):
+    """Test middleware loading from a JSON config file via --middleware-config."""
+
+    test_file = tmp_path / 'test.pdf'
+    test_file.write_text('dummy pdf content')
+
+    config_file = tmp_path / 'middleware.json'
+    config_file.write_text('["parxy_core.middleware.SimpleMiddleware"]')
+
+    with patch('parxy_cli.commands.parse.Parxy') as mock_parxy:
+        mock_parxy.default_driver.return_value = 'pymupdf'
+        mock_parxy.batch_iter.return_value = iter(
+            [
+                BatchResult(
+                    file=str(test_file),
+                    driver='pymupdf',
+                    document=mock_document,
+                    error=None,
+                )
+            ]
+        )
 
         result = runner.invoke(
             app,
-            [str(test_file), '--middleware-profile', 'default'],
+            [str(test_file), '--middleware-config', str(config_file)],
         )
 
         assert result.exit_code == 0
         mock_parxy.clear_middleware.assert_called_once()
-        mock_parxy.with_middleware.assert_not_called()
+        mock_parxy.with_middleware.assert_called_once_with(
+            ['parxy_core.middleware.SimpleMiddleware']
+        )
 
 
-def test_parse_command_with_unknown_middleware_profile_fails(
+def test_parse_command_with_middleware_config_missing_file_fails(
     runner, mock_document, tmp_path
 ):
-    """Test unknown middleware profile returns a CLI validation error."""
+    """Test that a missing middleware config file returns a CLI error."""
 
     test_file = tmp_path / 'test.pdf'
     test_file.write_text('dummy pdf content')
 
     with patch('parxy_cli.commands.parse.Parxy') as mock_parxy:
-        mock_parxy.parse.return_value = mock_document
         mock_parxy.default_driver.return_value = 'pymupdf'
 
         result = runner.invoke(
             app,
-            [str(test_file), '--middleware-profile', 'does_not_exist'],
+            [str(test_file), '--middleware-config', str(tmp_path / 'nonexistent.json')],
         )
 
         assert result.exit_code != 0
-        mock_parxy.parse.assert_not_called()
+        mock_parxy.batch_iter.assert_not_called()
 
 
 def test_collect_files_non_recursive(tmp_path):