Implement sdmx import harness using autoschematisation

requirements_all.txt

@@ -9,6 +9,7 @@
 
 absl-py
 chembl-webresource-client
+dataclasses-json
 deepdiff
 earthengine-api
 flask_restful

tools/agentic_import/backup_processor_run_test.py

@@ -45,6 +45,17 @@ def _read_manifest(self, backup_dir: Path) -> str:
         with open(manifest_path, 'r') as manifest_file:
             return manifest_file.read()
 
+    def test_absolute_path_copied(self):
+        absolute_file = self.working_dir / 'abs.txt'
+        absolute_file.write_text('absolute')
+
+        backup_dir = self._run_backup([str(absolute_file)])
+
+        self.assertTrue((backup_dir / 'abs.txt').exists())
+        manifest = self._read_manifest(backup_dir)
+        self.assertIn(str(absolute_file), manifest)
+        self.assertNotIn('Skipped (missing or blocked):', manifest)
+
     def test_copies_requested_files(self):
         first = self.working_dir / 'a.txt'
         second = self.working_dir / 'b.txt'

tools/agentic_import/pipeline.py

@@ -0,0 +1,136 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Generic building blocks for lightweight agentic pipelines.
+"""
+
+from __future__ import annotations
+
+import abc
+from dataclasses import dataclass
+from typing import Sequence
+
+from absl import logging
+
+
+class PipelineAbort(Exception):
+    """Raised when pipeline execution should stop early for any reason."""
+
+
+class Step(abc.ABC):
+    """Abstract pipeline step interface."""
+
+    @property
+    @abc.abstractmethod
+    def name(self) -> str:
+        """Human friendly identifier used for logging."""
+
+    @property
+    @abc.abstractmethod
+    def version(self) -> str:
+        """Version string used for invalidation decisions."""
+
+    @abc.abstractmethod
+    def run(self) -> None:
+        """Execute the step. Raise an exception to signal failure."""
+
+
+class BaseStep(Step, abc.ABC):
+    """Helper base class that stores mandatory metadata."""
+
+    def __init__(self, *, name: str, version: str) -> None:
+        if not name:
+            raise ValueError("step requires a name")
+        self._name = name
+        self._version = version
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def version(self) -> str:
+        return self._version
+
+
+@dataclass(frozen=True)
+class Pipeline:
+    steps: Sequence[Step]
+
+
+class PipelineCallback:
+    """Lifecycle hooks consumed by the runner; defaults are no-ops."""
+
+    def before_step(self, step: Step) -> None:
+        """Called immediately before `step.run()`; raising an error skips execution."""
+
+    def after_step(self, step: Step, *, error: Exception | None = None) -> None:
+        """Runs once per step after `step.run()` succeeds or raises."""
+
+
+class CompositeCallback(PipelineCallback):
+    """Fans out events to child callbacks in order."""
+
+    def __init__(self, callbacks: Sequence[PipelineCallback]) -> None:
+        self._callbacks = list(callbacks)
+
+    def before_step(self, step: Step) -> None:
+        for callback in self._callbacks:
+            callback.before_step(step)
+
+    def after_step(self, step: Step, *, error: Exception | None = None) -> None:
+        for callback in self._callbacks:
+            callback.after_step(step, error=error)
+
+
+@dataclass(frozen=True)
+class RunnerConfig:
+    """Placeholder for future runner toggles."""
+
+
+class PipelineRunner:
+
+    def __init__(self, config: RunnerConfig | None = None) -> None:
+        self._config = config or RunnerConfig()
+
+    def run(self,
+            pipeline: Pipeline,
+            callback: PipelineCallback | None = None) -> None:
+        current_step: Step | None = None
+        steps = pipeline.steps
+        logging.info(f"Starting pipeline with {len(steps)} steps")
+        try:
+            for step in steps:
+                current_step = step
+                logging.info(f"[STEP START] {step.name} (v{step.version})")
+                if callback:
+                    callback.before_step(step)
+                error: Exception | None = None
+                try:
+                    step.run()
+                except Exception as exc:  # pylint: disable=broad-except
+                    error = exc
+                    logging.exception(
+                        f"[STEP END] {step.name} (v{step.version}) status=failed"
+                    )
+                    raise
+                finally:
+                    if callback:
+                        callback.after_step(step, error=error)
+                logging.info(
+                    f"[STEP END] {step.name} (v{step.version}) status=succeeded"
+                )
+            logging.info("Pipeline completed")
+        except PipelineAbort:
+            name = current_step.name if current_step else "<none>"
+            logging.info(f"[STEP END] {name} status=aborted; pipeline aborted")

tools/agentic_import/pipeline_test.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for the agentic pipeline skeleton."""
+
+import os
+import sys
+import unittest
+
+_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(_SCRIPT_DIR)
+
+from pipeline import (  # pylint: disable=import-error
+    BaseStep, CompositeCallback, Pipeline, PipelineAbort, PipelineCallback,
+    PipelineRunner, RunnerConfig, Step,
+)
+
+
+class _TrackingStep(BaseStep):
+
+    def __init__(self, name: str, events: list[str]) -> None:
+        super().__init__(name=name, version="1")
+        self._events = events
+        self.executed = False
+
+    def run(self) -> None:
+        self.executed = True
+        self._events.append(f"run:{self.name}")
+
+
+class _FailingStep(BaseStep):
+
+    def __init__(self, *, name: str, version: str) -> None:
+        super().__init__(name=name, version=version)
+
+    def run(self) -> None:
+        raise ValueError("boom")
+
+
+class PipelineRunnerTest(unittest.TestCase):
+
+    def _build_pipeline(self, events: list[str]) -> Pipeline:
+        step_one = _TrackingStep("one", events)
+        step_two = _TrackingStep("two", events)
+        return Pipeline(steps=[step_one, step_two])
+
+    def test_on_before_step_runs_before_each_step(self) -> None:
+        events: list[str] = []
+
+        class RecordingCallback(PipelineCallback):
+
+            def before_step(self, step: Step) -> None:
+                events.append(f"before:{step.name}")
+
+        pipeline = self._build_pipeline(events)
+        PipelineRunner(RunnerConfig()).run(pipeline, RecordingCallback())
+
+        self.assertEqual(
+            events,
+            [
+                "before:one",
+                "run:one",
+                "before:two",
+                "run:two",
+            ],
+        )
+
+    def test_pipeline_abort_skips_downstream_steps(self) -> None:
+        events: list[str] = []
+        pipeline = self._build_pipeline(events)
+        runner = PipelineRunner(RunnerConfig())
+
+        class AbortOnSecond(PipelineCallback):
+
+            def before_step(self, step: Step) -> None:
+                if step.name == "two":
+                    raise PipelineAbort("stop")
+
+        runner.run(pipeline, AbortOnSecond())
+
+        self.assertEqual(events, ["run:one"])
+        # PipelineAbort is swallowed by the runner, so execution simply stops.
+
+    def test_before_step_exception_skips_after_step(self) -> None:
+        events: list[str] = []
+        pipeline = Pipeline(steps=[_TrackingStep("one", events)])
+        runner = PipelineRunner(RunnerConfig())
+
+        class RecordingCallback(PipelineCallback):
+
+            def before_step(self, step: Step) -> None:
+                events.append(f"before:{step.name}")
+                raise RuntimeError("boom")
+
+            def after_step(self,
+                           step: Step,
+                           *,
+                           error: Exception | None = None) -> None:
+                del step, error
+                events.append("after-called")
+
+        with self.assertRaises(RuntimeError):
+            runner.run(pipeline, RecordingCallback())
+
+        self.assertEqual(events, ["before:one"])
+
+    def test_after_step_receives_error_when_step_fails(self) -> None:
+
+        class RecordingCallback(PipelineCallback):
+
+            def __init__(self) -> None:
+                self.after_calls: list[tuple[str, str | None]] = []
+
+            def after_step(self,
+                           step: Step,
+                           *,
+                           error: Exception | None = None) -> None:
+                name = step.name
+                error_name = type(error).__name__ if error else None
+                self.after_calls.append((name, error_name))
+
+        callback = RecordingCallback()
+        pipeline = Pipeline(steps=[_FailingStep(name="fail-step", version="1")])
+
+        with self.assertRaises(ValueError):
+            PipelineRunner(RunnerConfig()).run(pipeline, callback)
+
+        self.assertEqual(callback.after_calls, [("fail-step", "ValueError")])
+
+
+class CompositeCallbackTest(unittest.TestCase):
+
+    def test_callbacks_run_in_order_for_each_hook(self) -> None:
+        events: list[str] = []
+
+        class RecordingCallback(PipelineCallback):
+
+            def __init__(self, label: str) -> None:
+                self._label = label
+
+            def before_step(self, step: Step) -> None:
+                events.append(f"{self._label}:before:{step.name}")
+
+            def after_step(self,
+                           step: Step,
+                           *,
+                           error: Exception | None = None) -> None:
+                del error
+                events.append(f"{self._label}:after:{step.name}")
+
+        composite = CompositeCallback(
+            [RecordingCallback("first"),
+             RecordingCallback("second")])
+        step = _TrackingStep("composite", events)
+
+        composite.before_step(step)
+        composite.after_step(step)
+
+        self.assertEqual(
+            events,
+            [
+                "first:before:composite",
+                "second:before:composite",
+                "first:after:composite",
+                "second:after:composite",
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()