ExecutionHooks is absorbed by TaskRuntimeBasePlugin and renamed to ExecutionHooksBasePlugin

Author: mexanick

src/dirac_cwl_proto/job/__init__.py

@@ -27,7 +27,7 @@
 from ruamel.yaml import YAML
 from schema_salad.exceptions import ValidationException
 
-from dirac_cwl_proto.metadata.core import TaskRuntimeBasePlugin
+from dirac_cwl_proto.metadata.core import ExecutionHooksBasePlugin
 from dirac_cwl_proto.submission_models import (
     JobParameterModel,
     JobSubmissionModel,
@@ -256,7 +256,7 @@ def submit_job_router(job: JobSubmissionModel) -> bool:
 def _pre_process(
     executable: CommandLineTool | Workflow | ExpressionTool,
     arguments: JobParameterModel | None,
-    runtime_metadata: TaskRuntimeBasePlugin | None,
+    runtime_metadata: ExecutionHooksBasePlugin | None,
     job_path: Path,
 ) -> list[str]:
     """
@@ -334,7 +334,7 @@ def _post_process(
     stdout: str,
     stderr: str,
     job_path: Path,
-    runtime_metadata: TaskRuntimeBasePlugin | None,
+    runtime_metadata: ExecutionHooksBasePlugin | None,
 ):
     """
     Post-process the job after execution.

src/dirac_cwl_proto/metadata/__init__.py

@@ -10,13 +10,11 @@
 from .core import (
     DataCatalogInterface,
     DataManager,
-    ExecutionHooks,
+    ExecutionHooksBasePlugin,
     SchedulingHint,
-    TaskRuntimeBasePlugin,
     TransformationDataManager,
 )
 from .registry import (
-    MetadataPluginRegistry,
     discover_plugins,
     get_registry,
 )
@@ -32,15 +30,12 @@
     pass
 
 __all__ = [
-    # Core classes
-    "TaskRuntimeBasePlugin",
-    "DataCatalogInterface",
+    # Core metadata and plugins
     "DataManager",
-    "ExecutionHooks",
-    "SchedulingHint",
     "TransformationDataManager",
-    # Registry functions
-    "MetadataPluginRegistry",
-    "discover_plugins",
+    "ExecutionHooksBasePlugin",
+    "SchedulingHint",
+    "DataCatalogInterface",
+    "MetadataRegistry",
     "get_registry",
 ]

src/dirac_cwl_proto/metadata/core.py

@@ -19,48 +19,6 @@
 T = TypeVar("T", bound="SchedulingHint")
 
 
-class ExecutionHooks(ABC):
-    """Abstract base class for execution hooks.
-
-    This class defines the interface for pre/post processing operations
-    that can be performed during job execution.
-    """
-
-    @abstractmethod
-    def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
-        """Pre-process job inputs and command.
-
-        Parameters
-        ----------
-        job_path : Path
-            Path to the job working directory.
-        command : List[str]
-            The command to be executed.
-
-        Returns
-        -------
-        List[str]
-            Modified command list.
-        """
-        pass
-
-    @abstractmethod
-    def post_process(self, job_path: Path) -> bool:
-        """Post-process job outputs.
-
-        Parameters
-        ----------
-        job_path : Path
-            Path to the job working directory.
-
-        Returns
-        -------
-        bool
-            True if post-processing succeeded, False otherwise.
-        """
-        pass
-
-
 class DataCatalogInterface(ABC):
     """Abstract interface for data catalog operations."""
 
@@ -120,8 +78,8 @@ def store_output(self, output_name: str, src_path: str) -> None:
         logger.info(f"Output {output_name} stored in {dest}")
 
 
-class TaskRuntimeBasePlugin(BaseModel, DataCatalogInterface, ExecutionHooks):
-    """Base class for all runtime plugin models.
+class ExecutionHooksBasePlugin(BaseModel, DataCatalogInterface):
+    """Base class for all runtime plugin models with execution hooks.
 
     This class combines Pydantic validation with the execution hooks
     and data catalog interfaces to provide a complete foundation for runtime plugin implementations.
@@ -151,12 +109,33 @@ def get_metadata_class(cls) -> str:
             name = name[:-8]  # Remove "Metadata" suffix
         return name
 
-    def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
-        """Default pre-processing: return command unchanged."""
+    def pre_process(
+        self, job_path: Path, command: List[str], **kwargs: Any
+    ) -> List[str]:
+        """Pre-process job inputs and command.
+
+        Parameters
+        ----------
+        job_path : Path
+            Path to the job working directory.
+        command : List[str]
+            The command to be executed.
+
+        Returns
+        -------
+        List[str]
+            Modified command list.
+        """
         return command
 
-    def post_process(self, job_path: Path) -> bool:
-        """Default post-processing: always succeed."""
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
+        """Post-process job outputs.
+
+        Parameters
+        ----------
+        job_path : Path
+            Path to the job working directory.
+        """
         return True
 
     def get_input_query(
@@ -284,11 +263,11 @@ def model_copy(
 
         return super().model_copy(update=merged_update, deep=deep)
 
-    def to_runtime(self, submitted: Optional[Any] = None) -> "TaskRuntimeBasePlugin":
+    def to_runtime(self, submitted: Optional[Any] = None) -> "ExecutionHooksBasePlugin":
         """
             Build and instantiate the runtime metadata implementation.
 
-        The returned object is an instance of :class:`TaskRuntimeBasePlugin` created
+        The returned object is an instance of :class:`ExecutionHooksBasePlugin` created
         by the metadata registry. The instantiation parameters are constructed
             by merging, in order:
 
@@ -308,7 +287,7 @@ def to_runtime(self, submitted: Optional[Any] = None) -> "TaskRuntimeBasePlugin"
 
             Returns
             -------
-            TaskRuntimeBasePlugin
+            ExecutionHooksBasePlugin
                 Runtime plugin implementation instantiated from the registry.
         """
         # Import here to avoid circular imports

src/dirac_cwl_proto/metadata/plugins/core.py

@@ -9,10 +9,12 @@
 from pathlib import Path
 from typing import Any, ClassVar, List, Optional, Union
 
-from ..core import TaskRuntimeBasePlugin
+from pydantic import Field
 
+from ..core import ExecutionHooksBasePlugin
 
-class UserMetadata(TaskRuntimeBasePlugin):
+
+class UserMetadata(ExecutionHooksBasePlugin):
     """Default user metadata model with no special processing.
 
     This is the simplest metadata model that performs no special input/output
@@ -22,7 +24,7 @@ class UserMetadata(TaskRuntimeBasePlugin):
     description: ClassVar[str] = "Basic user metadata with no special processing"
 
 
-class AdminMetadata(TaskRuntimeBasePlugin):
+class AdminMetadata(ExecutionHooksBasePlugin):
     """Administrative metadata model with enhanced logging.
 
     This metadata model provides additional logging and monitoring
@@ -44,34 +46,36 @@ class AdminMetadata(TaskRuntimeBasePlugin):
     enable_monitoring: bool = True
     admin_level: int = 1
 
-    def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
+    def pre_process(
+        self, job_path: Path, command: List[str], **kwargs: Any
+    ) -> List[str]:
         """Add logging configuration to command."""
         if self.log_level != "INFO":
             command.extend(["--log-level", self.log_level])
         return command
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Enhanced post-processing with monitoring."""
         if self.enable_monitoring:
             # Could send metrics to monitoring system
             pass
         return True
 
 
-class QueryBasedMetadata(TaskRuntimeBasePlugin):
+class QueryBasedMetadata(ExecutionHooksBasePlugin):
     """Metadata model that supports query-based input resolution.
 
     This model demonstrates how to implement query-based data discovery
     using metadata parameters.
     """
 
-    description: ClassVar[str] = "Metadata with query-based input resolution"
+    description: ClassVar[str] = "Query-based metadata for data discovery"
 
     # Query parameters
-    query_root: Optional[str] = None
-    site: Optional[str] = None
-    campaign: Optional[str] = None
-    data_type: Optional[str] = None
+    query_root: str = Field(default="/", description="Root path for queries")
+    site: Optional[str] = Field(default=None, description="Site to query")
+    campaign: Optional[str] = Field(default=None, description="Campaign name")
+    data_type: Optional[str] = Field(default=None, description="Data type")
 
     def get_input_query(
         self, input_name: str, **kwargs: Any
@@ -118,7 +122,7 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
         return base_path / "default"
 
 
-class TaskWithMetadataQueryPlugin(TaskRuntimeBasePlugin):
+class TaskWithMetadataQueryPlugin(ExecutionHooksBasePlugin):
     """Metadata plugin that demonstrates query-based input resolution.
 
     This class provides methods to query metadata and generate input paths

src/dirac_cwl_proto/metadata/plugins/gaussian.py

@@ -10,13 +10,14 @@
 from pathlib import Path
 from typing import Any, ClassVar, List, Optional, Union
 
-from ..core import TaskRuntimeBasePlugin
+from ..core import ExecutionHooksBasePlugin
 
 
-class DataGenerationMetadata(TaskRuntimeBasePlugin):
+class DataGenerationMetadata(ExecutionHooksBasePlugin):
     """Data generation metadata model for Gaussian fitting.
 
-    This model handles generation of test data for Gaussian fitting algorithms.
+    This model handles generation of test data for Gaussian fitting algorithms.:w
+
     It supports generating multiple output files with configurable names.
 
     Parameters
@@ -43,7 +44,7 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
 
         return base_path / "data-generation"
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Post process the generated data files."""
         success = False
 
@@ -71,7 +72,7 @@ def post_process(self, job_path: Path) -> bool:
         return success
 
 
-class GaussianFitMetadata(TaskRuntimeBasePlugin):
+class GaussianFitMetadata(ExecutionHooksBasePlugin):
     """Gaussian fitting metadata model.
 
     This model handles Gaussian fitting analysis on generated data sets.
@@ -85,7 +86,7 @@ class GaussianFitMetadata(TaskRuntimeBasePlugin):
         Second set of input data files.
     """
 
-    description: ClassVar[str] = "Gaussian fitting analysis on data sets"
+    description: ClassVar[str] = "Gaussian fitting analysis for generated data"
 
     # Input data
     data1: Optional[List] = None
@@ -110,7 +111,7 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
             return Path("filecatalog") / "gaussian_fit" / "fit"
         return None
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Post process the fitting results."""
         outputs = glob.glob(str(job_path / "fit*"))
         if outputs:

src/dirac_cwl_proto/metadata/plugins/lhcb.py

@@ -17,10 +17,10 @@
 from pydantic import Field
 from ruamel.yaml import YAML
 
-from ..core import TaskRuntimeBasePlugin
+from ..core import ExecutionHooksBasePlugin
 
 
-class LHCbMetadata(TaskRuntimeBasePlugin):
+class LHCbMetadata(ExecutionHooksBasePlugin):
     """Base metadata model for LHCb experiment.
 
     This class provides common functionality for all LHCb metadata models,
@@ -84,7 +84,9 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
 
         return base / "outputs"
 
-    def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
+    def pre_process(
+        self, job_path: Path, command: List[str], **kwargs: Any
+    ) -> List[str]:
         """Pre-process LHCb simulation job.
 
         This method calculates the optimal number of events to simulate
@@ -170,7 +172,7 @@ def _update_job_parameters(self, job_path: Path, command: List[str]) -> None:
         with open(parameters_path, "w") as f:
             YAML().dump(parameter_dict, f)
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Post-process LHCb simulation outputs."""
         success = True
 
@@ -248,7 +250,9 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
 
         return base / "outputs"
 
-    def pre_process(self, job_dir: Path, command: List[str]) -> List[str]:
+    def pre_process(
+        self, job_path: Path, command: List[str], **kwargs: Any
+    ) -> List[str]:
         """Pre-process the job command for reconstruction."""
         # Only add LHCb-specific arguments if this is not a CWL workflow execution
         # (i.e., when running LHCb applications directly, not cwltool)
@@ -263,7 +267,7 @@ def pre_process(self, job_dir: Path, command: List[str]) -> List[str]:
 
         return command
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Post-process LHCb reconstruction outputs."""
         success = True
 
@@ -355,7 +359,9 @@ def get_output_query(self, output_name: str) -> Optional[Path]:
 
         return base / "results"
 
-    def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
+    def pre_process(
+        self, job_path: Path, command: List[str], **kwargs: Any
+    ) -> List[str]:
         """Pre-process LHCb analysis job."""
         # Add analysis-specific parameters
         command.extend(["--analysis", self.analysis_name])
@@ -369,7 +375,7 @@ def pre_process(self, job_path: Path, command: List[str]) -> List[str]:
 
         return command
 
-    def post_process(self, job_path: Path) -> bool:
+    def post_process(self, job_path: Path, **kwargs: Any) -> bool:
         """Post-process LHCb analysis outputs."""
         success = True