Add TransformContext to TransformMetadata

Enriches all transform types (Python, Node, plugin) with a TransformContext
carrying bblock metadata, example/snippet data, file paths, and register
configuration. Subprocess transforms receive a serialized copy via CLI arg
(Python) or injected JSON (Node/plugin), deserialized to a SimpleNamespace
for consistent attribute access across all transform types.

Author: avillar

ogc/bblocks/models.py

@@ -792,6 +792,37 @@ def get_url(self, path: str | Path) -> str:
         return f"{self.base_url}{os.path.relpath(Path(path).resolve(), self._cwd)}"
 
 
+@dataclasses.dataclass
+class TransformContext:
+    bblock_id: str
+    bblock_name: str | None
+    bblock_version: str | None
+    bblock_tags: list
+    bblock_files_path: str | None
+    bblock_annotated_path: str | None
+    bblock_metadata: dict
+    example_index: int
+    example: dict
+    snippet_index: int
+    snippet: dict
+    output_file: str | None
+    output_dir: str | None
+    working_dir: str
+    source_schema_path: str | None = None
+    annotated_schema_path: str | None = None
+    jsonld_context_path: str | None = None
+    shacl_shapes_paths: list = dataclasses.field(default_factory=list)
+    base_url: str | None = None
+    github_base_url: str | None = None
+    git_repository: str | None = None
+    id_prefix: str = ''
+    imported_register_urls: list = dataclasses.field(default_factory=list)
+    transform_plugins: list = dataclasses.field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return json.loads(json.dumps(dataclasses.asdict(self), default=str))
+
+
 @dataclasses.dataclass
 class TransformMetadata:
     type: str
@@ -802,6 +833,7 @@ class TransformMetadata:
     metadata: dict | None = None
     sandbox_dir: Path | None = None
     id: str | None = None
+    ctx: TransformContext | None = None
 
 
 @dataclasses.dataclass

ogc/bblocks/postprocess.py

@@ -211,7 +211,12 @@ def do_postprocess(bblock: BuildingBlock, light: bool = False) -> bool:
         if not light and (not steps or 'transforms' in steps) and bblock.transforms:
             logger.info("Running transforms")
             apply_transforms(bblock, outputs_path=test_outputs_path, base_url=base_url,
-                             sandbox_dir=sandbox_dir, bblocks_register=bbr)
+                             sandbox_dir=sandbox_dir, bblocks_register=bbr,
+                             github_base_url=github_base_url,
+                             git_repository=additional_metadata.get('gitRepository'),
+                             id_prefix=id_prefix,
+                             imported_register_urls=imported_registers,
+                             transform_plugins=transform_plugins)
 
         if bblock.examples:
             for example in bblock.examples:

ogc/bblocks/transform.py

@@ -21,14 +21,24 @@
 from ogc.bblocks.sandbox import ensure_venv, venv_needs_recreate
 
 from ogc.bblocks import mimetypes
-from ogc.bblocks.models import BuildingBlock, BuildingBlockRegister, ImportedBBlockProxy, TransformMetadata, TransformResult, BuildingBlockError
+from ogc.bblocks.models import BuildingBlock, BuildingBlockRegister, ImportedBBlockProxy, TransformContext, TransformMetadata, TransformResult, BuildingBlockError
 from ogc.bblocks.transformers import transformers
 from ogc.bblocks.util import sanitize_filename
 from ogc.bblocks.validate import validate_transform_output, report_to_dict
 
 _SUBPROCESS_TRANSFORM_TYPES = ('python', 'node')
 
 
+def _rel(path: Path | str | None, cwd: Path) -> str | None:
+    if path is None:
+        return None
+    p = Path(path)
+    try:
+        return str(p.relative_to(cwd))
+    except ValueError:
+        return str(p)
+
+
 def _pip_to_url(pip_spec: str) -> str | None:
     """Derive a human-facing URL from a pip install specifier, or None if not applicable."""
     if not pip_spec:
@@ -230,7 +240,12 @@ def apply_transforms(bblock: BuildingBlock,
                      output_subpath='transforms',
                      base_url: str | None = None,
                      sandbox_dir: Path | None = None,
-                     bblocks_register: BuildingBlockRegister | None = None):
+                     bblocks_register: BuildingBlockRegister | None = None,
+                     github_base_url: str | None = None,
+                     git_repository: str | None = None,
+                     id_prefix: str = '',
+                     imported_register_urls: list[str] | None = None,
+                     transform_plugins: list[dict] | None = None):
 
     if not bblock.transforms:
         return
@@ -330,6 +345,38 @@ def apply_transforms(bblock: BuildingBlock,
                 if example_prefixes:
                     metadata['_prefixes'] = example_prefixes
 
+                ctx = TransformContext(
+                    bblock_id=bblock.identifier,
+                    bblock_name=bblock.name,
+                    bblock_version=bblock.version,
+                    bblock_tags=list(bblock.metadata.get('tags') or []),
+                    bblock_files_path=_rel(bblock.files_path, cwd),
+                    bblock_annotated_path=_rel(bblock.annotated_path, cwd),
+                    bblock_metadata=bblock.metadata,
+                    example_index=example_id,
+                    example={k: v for k, v in example.items() if k != 'snippets'},
+                    snippet_index=snippet_id,
+                    snippet={k: v for k, v in snippet.items() if k != 'code'},
+                    output_file=_rel(output_fn, cwd),
+                    output_dir=_rel(output_dir, cwd),
+                    working_dir=str(cwd),
+                    source_schema_path=(
+                        _rel(bblock.schema.path, cwd) if bblock.schema.is_path else bblock.schema.url
+                    ) if bblock.schema else None,
+                    annotated_schema_path=_rel(bblock.annotated_schema, cwd) if bblock.annotated_schema.is_file() else None,
+                    jsonld_context_path=_rel(bblock.jsonld_context, cwd) if bblock.jsonld_context.is_file() else None,
+                    shacl_shapes_paths=[
+                        s if isinstance(s, str) else _rel(s, cwd)
+                        for s in (bblock.shacl_shapes or [])
+                    ],
+                    base_url=base_url,
+                    github_base_url=github_base_url,
+                    git_repository=git_repository,
+                    id_prefix=id_prefix,
+                    imported_register_urls=list(imported_register_urls or []),
+                    transform_plugins=list(transform_plugins or []),
+                )
+
                 transform_metadata = TransformMetadata(type=transform['type'],
                                                        id=transform['id'],
                                                        source_mime_type=snippet_mime_type,
@@ -338,7 +385,8 @@ def apply_transforms(bblock: BuildingBlock,
                                                        metadata=metadata,
                                                        input_data=snippet['code'],
                                                        sandbox_dir=transform_sandboxes.get(
-                                                           transform['id'], sandbox_dir))
+                                                           transform['id'], sandbox_dir),
+                                                       ctx=ctx)
 
                 try:
                     result = transformer.transform(transform_metadata)

ogc/bblocks/transformers/_plugin_harness.py

@@ -18,18 +18,20 @@
       {"success": false, "output": null,        "binary": false, "stderr": "<str>"}
 
     The metadata object passed to transform() has these attributes:
-      type             (str)   transform type identifier
-      transform_content (str)  code/script declared in transforms.yaml
-      input_data       (str)   example snippet text
+      type             (str)       transform type identifier
+      transform_content (str)      code/script declared in transforms.yaml
+      input_data       (str)       example snippet text
       source_mime_type (str)
       target_mime_type (str)
-      metadata         (dict)  extra metadata (keys starting with _ excluded)
-      sandbox_dir      None    always None in subprocess context
+      metadata         (dict)      extra metadata (keys starting with _ excluded)
+      sandbox_dir      None        always None in subprocess context
+      ctx              SimpleNamespace | None  transform context (bblock, example, register info)
 """
 import importlib
 import inspect
 import json
 import sys
+import types
 from base64 import b64encode
 
 
@@ -51,7 +53,7 @@ def _transformer_classes(module):
 class _Meta:
     """Minimal TransformMetadata-compatible namespace passed to plugin transform()."""
     __slots__ = ('type', 'transform_content', 'input_data',
-                 'source_mime_type', 'target_mime_type', 'metadata', 'sandbox_dir')
+                 'source_mime_type', 'target_mime_type', 'metadata', 'sandbox_dir', 'ctx')
 
 
 # ---------------------------------------------------------------------------
@@ -87,6 +89,8 @@ def _transform(meta_json: str) -> None:
     m.metadata = meta_dict.get('metadata', {})
     m.input_data = sys.stdin.buffer.read().decode('utf-8')
     m.sandbox_dir = None
+    ctx_dict = meta_dict.get('context') or {}
+    m.ctx = types.SimpleNamespace(**ctx_dict) if ctx_dict else None
 
     module_path = meta_dict['module']
     transform_type = meta_dict['type']

ogc/bblocks/transformers/node.py

@@ -47,6 +47,7 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
             'targetMimeType': metadata.target_mime_type,
             'metadata': {k: v for k, v in (metadata.metadata or {}).items()
                          if not k.startswith('_')},
+            'context': metadata.ctx.to_dict() if metadata.ctx else None,
         }
 
         harness = f"""\

ogc/bblocks/transformers/plugin.py

@@ -83,6 +83,7 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
             'target_mime_type': metadata.target_mime_type,
             'metadata': {k: v for k, v in (metadata.metadata or {}).items()
                          if not k.startswith('_')},
+            'context': metadata.ctx.to_dict() if metadata.ctx else None,
         }
 
         result = subprocess.run(

ogc/bblocks/transformers/python.py

@@ -62,11 +62,15 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
             'target_mime_type': metadata.target_mime_type,
             'metadata': {k: v for k, v in (metadata.metadata or {}).items()
                          if not k.startswith('_')},
+            'context': metadata.ctx.to_dict() if metadata.ctx else None,
         }
 
         harness = f"""\
-import sys as _sys
-transform_metadata = {json.dumps(transform_metadata_dict)}
+import sys as _sys, json as _json, types as _types
+_d = _json.loads(_sys.argv[1])
+if isinstance(_d.get('context'), dict):
+    _d['context'] = _types.SimpleNamespace(**_d['context'])
+transform_metadata = _types.SimpleNamespace(**_d)
 input_data = _sys.stdin.read()
 output_data = None
 _real_stdout = _sys.stdout
@@ -83,7 +87,7 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
 
         try:
             result = subprocess.run(
-                [str(python_bin), harness_path],
+                [str(python_bin), harness_path, json.dumps(transform_metadata_dict)],
                 input=metadata.input_data.encode('utf-8') if isinstance(metadata.input_data, str) else metadata.input_data,
                 capture_output=True,
             )