Hints serverside

src/dirac_cwl_proto/execution_hooks/core.py

@@ -9,7 +9,7 @@
 import logging
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Any, ClassVar, Dict, List, Mapping, Optional, TypeVar, Union
+from typing import Any, ClassVar, Dict, List, Mapping, Optional, Self, TypeVar, Union
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
 
@@ -273,7 +273,7 @@ def model_copy(
         update: Optional[Mapping[str, Any]] = None,
         *,
         deep: bool = False,
-    ) -> "ExecutionHooksHint":
+    ) -> Self:
         """Enhanced model copy with intelligent merging of dict fields (including configuration)."""
         if update is None:
             update = {}
@@ -355,7 +355,7 @@ def _dash_to_snake(s: str) -> str:
         return get_registry().instantiate_plugin(descriptor)
 
     @classmethod
-    def from_cwl(cls, cwl_object: Any) -> "ExecutionHooksHint":
+    def from_cwl(cls, cwl_object: Any) -> Self:
         """Extract metadata descriptor from CWL object using Hint interface."""
         descriptor = cls()
         hints = getattr(cwl_object, "hints", []) or []

src/dirac_cwl_proto/job/__init__.py

@@ -27,7 +27,10 @@
 from ruamel.yaml import YAML
 from schema_salad.exceptions import ValidationException
 
-from dirac_cwl_proto.execution_hooks.core import ExecutionHooksBasePlugin
+from dirac_cwl_proto.execution_hooks.core import (
+    ExecutionHooksBasePlugin,
+    ExecutionHooksHint,
+)
 from dirac_cwl_proto.submission_models import (
     JobInputModel,
     JobSubmissionModel,
@@ -126,12 +129,7 @@ def submit_job_client(
                 f"\t[green]:heavy_check_mark:[/green] Parameter {parameter_p}"
             )
 
-    job = JobSubmissionModel(
-        task=task,
-        parameters=parameters,
-        scheduling=job_scheduling,
-        execution_hooks=job_metadata,
-    )
+    job = JobSubmissionModel(task=task, parameters=parameters)
     console.print(
         "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Job(s) validated."
     )
@@ -230,8 +228,6 @@ def submit_job_router(job: JobSubmissionModel) -> bool:
                 JobSubmissionModel(
                     task=job.task,
                     parameters=[parameter],
-                    scheduling=job.scheduling,
-                    execution_hooks=job.execution_hooks,
                 )
             )
     logger.info("Job(s) validated!")
@@ -364,8 +360,10 @@ def run_job(job: JobSubmissionModel) -> bool:
     logger = logging.getLogger("JobWrapper")
     # Instantiate runtime metadata from the serializable descriptor and
     # the job context so implementations can access task inputs/overrides.
+    job_execution_hooks = ExecutionHooksHint.from_cwl(job.task)
+
     runtime_metadata = (
-        job.execution_hooks.to_runtime(job) if job.execution_hooks else None
+        job_execution_hooks.to_runtime(job) if job_execution_hooks else None
     )
 
     # Isolate the job in a specific directory

src/dirac_cwl_proto/production/__init__.py

@@ -18,13 +18,8 @@
 )
 from rich import print_json
 from rich.console import Console
-from ruamel.yaml import YAML
 from schema_salad.exceptions import ValidationException
 
-from dirac_cwl_proto.execution_hooks import (
-    SchedulingHint,
-    TransformationExecutionHooksHint,
-)
 from dirac_cwl_proto.submission_models import (
     ProductionSubmissionModel,
     TransformationSubmissionModel,
@@ -45,10 +40,6 @@
 @app.command("submit")
 def submit_production_client(
     task_path: str = typer.Argument(..., help="Path to the CWL file"),
-    steps_metadata_path: str = typer.Option(
-        None,
-        help="Path to metadata file used to generate the transformations (one entry per step)",
-    ),
     # Specific parameter for the purpose of the prototype
     local: Optional[bool] = typer.Option(
         True, help="Run the job locally instead of submitting it to the router"
@@ -78,33 +69,10 @@ def submit_production_client(
         )
         return typer.Exit(code=1)
     console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}")
-
-    # Load the metadata: at this stage, only the structure is validated, not the content
-    steps_metadata = {}
-    if steps_metadata_path:
-        with open(steps_metadata_path, "r") as file:
-            steps_metadata = YAML(typ="safe").load(file)
-
-    production_step_execution_hooks = {}
-    production_step_scheduling = {}
-    for step_name, step_data in steps_metadata.items():
-        # Extract metadata and scheduling from step_data
-        metadata_config = step_data.get("execution-hooks", {})
-        scheduling_config = step_data.get("scheduling", {})
-
-        # Create TransformationExecutionHooksHint with the metadata
-        production_step_execution_hooks[step_name] = TransformationExecutionHooksHint(
-            **metadata_config
-        )
-        production_step_scheduling[step_name] = SchedulingHint(**scheduling_config)
     console.print("\t[green]:heavy_check_mark:[/green] Metadata")
 
     # Create the production
-    transformation = ProductionSubmissionModel(
-        task=task,
-        steps_execution_hooks=production_step_execution_hooks,
-        steps_scheduling=production_step_scheduling,
-    )
+    production = ProductionSubmissionModel(task=task)
     console.print(
         "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Production validated."
     )
@@ -113,8 +81,8 @@ def submit_production_client(
     console.print(
         "[blue]:information_source:[/blue] [bold]CLI:[/bold] Submitting the production..."
     )
-    print_json(transformation.model_dump_json(indent=4))
-    if not submit_production_router(transformation):
+    print_json(production.model_dump_json(indent=4))
+    if not submit_production_router(production):
         console.print(
             "[red]:heavy_multiplication_x:[/red] [bold]CLI:[/bold] Failed to run production."
         )
@@ -170,31 +138,32 @@ def _get_transformations(
     """
     # Create a subworkflow and a transformation for each step
     transformations = []
+    configuration = _get_configuration(production.task)
+
     for step in production.task.steps:
         step_task = _create_subworkflow(
             step, str(production.task.cwlVersion), production.task.inputs
         )
-        configuration = _get_configuration(production.task)
+        if step_task.hints is None:
+            step_task.hints = []
 
         # Get the execution_hooks & description for the step
         step_id = step.id.split("#")[-1]
-        step_data: TransformationExecutionHooksHint = (
-            production.steps_execution_hooks.get(
-                step_id,
-                TransformationExecutionHooksHint(),
-            )
-        )
-        step_scheduling: SchedulingHint = production.steps_scheduling.get(
-            step_id,
-            SchedulingHint(),
-        )
-        step_data.configuration.update(configuration)
+        if production.task.hints is not None:
+            for hint in production.task.hints:
+                if hint["class"] == "$namespaces":
+                    step_task.hints.append(hint)
+                if hint["class"] == step_id:
+                    execution_hook_hint = {
+                        **hint.pop("dirac:execution-hooks"),
+                        "class": "dirac:execution-hooks",
+                        "configuration": configuration,
+                    }
+                    step_task.hints.append(execution_hook_hint)
 
         transformations.append(
             TransformationSubmissionModel(
                 task=step_task,
-                execution_hooks=step_data,
-                scheduling=step_scheduling,
             )
         )
     return transformations

src/dirac_cwl_proto/submission_models.py

@@ -20,7 +20,6 @@
 from dirac_cwl_proto.execution_hooks import (
     ExecutionHooksHint,
     SchedulingHint,
-    TransformationExecutionHooksHint,
 )
 
 # -----------------------------------------------------------------------------
@@ -50,8 +49,6 @@ class JobSubmissionModel(BaseModel):
 
     task: CommandLineTool | Workflow | ExpressionTool
     parameters: list[JobInputModel] | None = None
-    scheduling: SchedulingHint
-    execution_hooks: ExecutionHooksHint
 
     @field_serializer("task")
     def serialize_task(self, value):
@@ -73,8 +70,6 @@ class TransformationSubmissionModel(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     task: CommandLineTool | Workflow | ExpressionTool
-    execution_hooks: TransformationExecutionHooksHint
-    scheduling: SchedulingHint
 
     @field_serializer("task")
     def serialize_task(self, value):
@@ -96,20 +91,18 @@ class ProductionSubmissionModel(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     task: Workflow
-    # Key: step name, Value: description & execution_hooks of a transformation
-    steps_execution_hooks: dict[str, TransformationExecutionHooksHint]
-    # Key: step name, Value: scheduling configuration for a transformation
-    steps_scheduling: dict[str, SchedulingHint] = {}
 
     @model_validator(mode="before")
     def validate_steps_metadata(cls, values):
         task = values.get("task")
-        steps_execution_hooks = values.get("steps_execution_hooks")
-
-        if task and steps_execution_hooks:
+        if task and task.hints:
             # Extract the available steps in the task
-            task_steps = set([step.id.split("#")[-1] for step in task.steps])
-            metadata_keys = set(steps_execution_hooks.keys())
+            task_steps = {step.id.split("#")[-1] for step in task.steps}
+            metadata_keys = {
+                hint["class"]
+                for hint in task.hints
+                if "class" in hint and hint["class"] != "$namespaces"
+            }
 
             # Check if all metadata keys exist in the task's workflow steps
             missing_steps = metadata_keys - task_steps

src/dirac_cwl_proto/transformation/__init__.py

@@ -14,11 +14,9 @@
 from cwl_utils.parser.cwl_v1_2 import File
 from rich import print_json
 from rich.console import Console
-from ruamel.yaml import YAML
 from schema_salad.exceptions import ValidationException
 
 from dirac_cwl_proto.execution_hooks import (
-    SchedulingHint,
     TransformationExecutionHooksHint,
 )
 from dirac_cwl_proto.job import submit_job_router
@@ -40,9 +38,6 @@
 @app.command("submit")
 def submit_transformation_client(
     task_path: str = typer.Argument(..., help="Path to the CWL file"),
-    metadata_path: Optional[str] = typer.Option(
-        None, help="Path to metadata file used to generate the input query"
-    ),
     # Specific parameter for the purpose of the prototype
     local: Optional[bool] = typer.Option(
         True, help="Run the jobs locally instead of submitting them to the router"
@@ -73,22 +68,7 @@ def submit_transformation_client(
         return typer.Exit(code=1)
     console.print(f"\t[green]:heavy_check_mark:[/green] Task {task_path}")
 
-    # Load the metadata: at this stage, only the structure is validated, not the content
-    metadata_model = TransformationExecutionHooksHint()
-    if metadata_path:
-        with open(metadata_path, "r") as file:
-            metadata = YAML(typ="safe").load(file)
-        metadata_model = TransformationExecutionHooksHint(**metadata)
-    console.print("\t[green]:heavy_check_mark:[/green] Metadata")
-
-    transformation_scheduling = SchedulingHint.from_cwl(task)
-    console.print("\t[green]:heavy_check_mark:[/green] Description")
-
-    transformation = TransformationSubmissionModel(
-        task=task,
-        execution_hooks=metadata_model,
-        scheduling=transformation_scheduling,
-    )
+    transformation = TransformationSubmissionModel(task=task)
     console.print(
         "[green]:heavy_check_mark:[/green] [bold]CLI:[/bold] Transformation validated."
     )
@@ -121,7 +101,7 @@ def submit_transformation_router(transformation: TransformationSubmissionModel)
 
     :param transformation: The transformation to start
 
-    :return: True if the transformation executed successfully, False otherwise
+    :return: True if the transformation is executed successfully, False otherwise
     """
     logger = logging.getLogger("TransformationRouter")
 
@@ -134,20 +114,28 @@ def submit_transformation_router(transformation: TransformationSubmissionModel)
     # - if there is no execution_hooks, the transformation is not waiting for an input and can go on
     # - if there is execution_hooks, the transformation is waiting for an input
     job_model_params = []
+
+    try:
+        transformation_execution_hooks = TransformationExecutionHooksHint.from_cwl(
+            transformation.task
+        )
+    except Exception as exc:
+        raise ValueError(f"Invalid DIRAC hints:\n{exc}") from exc
+
     if (
-        transformation.execution_hooks.configuration
-        and transformation.execution_hooks.group_size
+        transformation_execution_hooks.configuration
+        and transformation_execution_hooks.group_size
     ):
         # Get the metadata class
-        transformation_metadata = transformation.execution_hooks.to_runtime(
+        transformation_metadata = transformation_execution_hooks.to_runtime(
             transformation
         )
 
         # Build the input cwl for the jobs to submit
         logger.info("Getting the input data for the transformation...")
         input_data_dict = {}
         min_length = None
-        for input_name, group_size in transformation.execution_hooks.group_size.items():
+        for input_name, group_size in transformation_execution_hooks.group_size.items():
             # Get input query
             logger.info(f"\t- Getting input query for {input_name}...")
             input_query = transformation_metadata.get_input_query(input_name)
@@ -177,8 +165,6 @@ def submit_transformation_router(transformation: TransformationSubmissionModel)
     jobs = JobSubmissionModel(
         task=transformation.task,
         parameters=job_model_params,
-        scheduling=transformation.scheduling,
-        execution_hooks=transformation.execution_hooks,
     )
     logger.info("Jobs built!")