Improve subprocess transform robustness and add binary + metadata dict support

- Python transforms: fix JSONDecodeError escaping _send_raw uncaught, add
  capture_output=True to pip/npm calls inside get_transformer (stdout pipe
  was being polluted), include full traceback when exception escapes
  transformer.transform()
- Node transforms: switch to structured JSON protocol (same as Python) so
  binary input/output is handled explicitly; fix binary input (was always
  read as utf-8); exceptions are now caught and surfaced with a stack trace;
  console.log no longer interferes with the protocol
- Plugin transforms: fix exception traceback (was str(e) only)
- All transforms: transform_metadata.metadata is now an _MNS instance —
  a SimpleNamespace-style object that also supports dict-style access
  (__getitem__, keys/values/items, 'in', iteration) so transform code can
  use either style interchangeably

Author: avillar

ogc/bblocks/transform.py

@@ -8,6 +8,7 @@
 import os.path
 import subprocess
 import sys
+import traceback
 
 logger = logging.getLogger(__name__)
 from packaging.version import Version
@@ -547,7 +548,7 @@ def apply_transforms(bblock: BuildingBlock,
                     logger.debug("Transform %s result: %s", transform['id'], result)
                 except Exception as e:
                     result = TransformResult(output=None, success=False,
-                                             stderr=f"{type(e).__name__}: {e}")
+                                             stderr=f"{type(e).__name__}: {e}\n{traceback.format_exc()}")
 
                 entry = {'success': result.success}
                 if not result.success:

ogc/bblocks/transformers/_plugin_harness.py

@@ -32,10 +32,26 @@
 import io
 import json
 import sys
+import traceback
 import types
 from base64 import b64encode
 
 
+class _MNS:
+    def __init__(self, **kw): self.__dict__.update(kw)
+    def __getitem__(self, k): return self.__dict__[k]
+    def __setitem__(self, k, v): self.__dict__[k] = v
+    def __delitem__(self, k): del self.__dict__[k]
+    def __contains__(self, k): return k in self.__dict__
+    def __iter__(self): return iter(self.__dict__)
+    def __len__(self): return len(self.__dict__)
+    def keys(self): return self.__dict__.keys()
+    def values(self): return self.__dict__.values()
+    def items(self): return self.__dict__.items()
+    def get(self, k, d=None): return self.__dict__.get(k, d)
+    def __repr__(self): return 'namespace(' + ', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()) + ')'
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -87,7 +103,7 @@ def _transform(meta_json: str) -> None:
     m.transform_content = meta_dict['transform_content']
     m.source_mime_type = meta_dict['source_mime_type']
     m.target_mime_type = meta_dict['target_mime_type']
-    m.metadata = meta_dict.get('metadata', {})
+    m.metadata = _MNS(**meta_dict.get('metadata', {}))
     raw = sys.stdin.buffer.read()
     m.input_data = raw if meta_dict.get('input_binary') else raw.decode('utf-8')
     m.sandbox_dir = None
@@ -117,9 +133,9 @@ def _transform(meta_json: str) -> None:
     try:
         result = transformer.transform(m)
         error = None
-    except Exception as e:
+    except Exception:
         result = None
-        error = str(e)
+        error = traceback.format_exc()
     finally:
         sys.stdout, sys.stderr = old_stdout, old_stderr
 

ogc/bblocks/transformers/node.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
+import base64
 import json
 import logging
 import os
@@ -19,17 +20,6 @@
 transform_type = 'node'
 
 
-def _decode_output(raw: bytes) -> tuple[str | bytes | None, bool]:
-    if not raw:
-        return None, False
-    try:
-        text = raw.decode('utf-8')
-        if '\x00' in text:
-            return raw, True
-        return text, False
-    except UnicodeDecodeError:
-        return raw, True
-
 
 class NodeTransformer(Transformer):
 
@@ -63,7 +53,8 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
 const path = require('path');
 const {{spawnSync}} = require('child_process');
 const transformMetadata = {json.dumps(transform_metadata_dict)};
-const inputData = fs.readFileSync(0, 'utf8');
+const _inputBuf = fs.readFileSync(0);
+const inputData = {json.dumps(isinstance(metadata.input_data, bytes))} ? _inputBuf : _inputBuf.toString('utf8');
 let outputData = null;
 
 const _TRANSFORMS_REGISTRY = {json.dumps(transforms_registry)};
@@ -179,14 +170,38 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
 }}
 
 const _origStdoutWrite = process.stdout.write.bind(process.stdout);
-process.stdout.write = process.stderr.write.bind(process.stderr);
-
-{metadata.transform_content}
+const _origStderrWrite = process.stderr.write.bind(process.stderr);
+const _captured = [];
+const _captureWrite = (chunk) => {{ _captured.push(Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk)); return true; }};
+process.stdout.write = _captureWrite;
+process.stderr.write = _captureWrite;
+
+let _success = true, _errMsg = null;
+try {{
+    {metadata.transform_content}
+}} catch (_e) {{
+    _success = false;
+    _errMsg = (_e && _e.stack) ? _e.stack : String(_e);
+}}
 
 process.stdout.write = _origStdoutWrite;
-if (outputData !== null) {{
-    process.stdout.write(typeof outputData === 'string' ? outputData : Buffer.from(outputData));
+process.stderr.write = _origStderrWrite;
+
+const _capturedStr = _captured.join('') || null;
+const _stderrStr = (_capturedStr !== null || _errMsg !== null)
+    ? (_capturedStr || '') + (_errMsg ? (_capturedStr ? '\\n' : '') + _errMsg : '')
+    : null;
+
+let _outB64 = null, _binary = false;
+if (_success && outputData !== null) {{
+    const _outBuf = Buffer.isBuffer(outputData)
+        ? outputData
+        : Buffer.from(typeof outputData === 'string' ? outputData : String(outputData), 'utf8');
+    _outB64 = _outBuf.toString('base64');
+    _binary = Buffer.isBuffer(outputData);
 }}
+
+_origStdoutWrite(JSON.stringify({{output: _outB64, binary: _binary, success: _success, stderr: _stderrStr}}) + '\\n');
 """
 
         with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False) as f:
@@ -208,15 +223,40 @@ def transform(self, metadata: TransformMetadata) -> TransformResult:
         finally:
             Path(harness_path).unlink(missing_ok=True)
 
-        stderr = result.stderr.decode('utf-8', errors='replace').replace(harness_path, '<transform>') or None
-        if stderr:
-            label = metadata.id or 'node'
+        label = metadata.id or 'node'
+
+        if not result.stdout.strip():
+            stderr = result.stderr.decode('utf-8', errors='replace').replace(harness_path, '<transform>') or 'Node transform produced no output'
             with log_indent():
                 for line in stderr.splitlines():
                     if line.strip():
                         logger.info("[%s] %s", label, line)
-        if result.returncode != 0:
             return TransformResult(output=None, success=False, stderr=stderr)
 
-        output, binary = _decode_output(result.stdout)
-        return TransformResult(output=output, success=True, stderr=stderr, binary=binary)
+        try:
+            data = json.loads(result.stdout)
+        except json.JSONDecodeError as e:
+            stderr = result.stderr.decode('utf-8', errors='replace').replace(harness_path, '<transform>') or f'Invalid JSON from Node harness: {e}'
+            return TransformResult(output=None, success=False, stderr=stderr)
+
+        stderr = data.get('stderr') or None
+        if stderr:
+            stderr = stderr.replace(harness_path, '<transform>')
+            with log_indent():
+                for line in stderr.splitlines():
+                    if line.strip():
+                        logger.info("[%s] %s", label, line)
+
+        output_b64 = data.get('output')
+        if output_b64 is not None:
+            output_bytes = base64.b64decode(output_b64)
+            output: str | bytes | None = output_bytes if data.get('binary') else output_bytes.decode('utf-8')
+        else:
+            output = None
+
+        return TransformResult(
+            output=output,
+            success=data.get('success', False),
+            stderr=stderr,
+            binary=bool(data.get('binary')),
+        )

ogc/bblocks/transformers/python.py

@@ -37,6 +37,25 @@
 # When dispatching non-Python transforms, the child call stack is propagated both via
 # the _BBLOCKS_CALL_STACK env var (read by Node's getTransformer) and via the request's
 # 'call_stack' field (read by the one-shot Python dispatch harness).
+# Class injected into every harness context.
+# Behaves like SimpleNamespace (attribute access) and like a dict (item access, keys/values/items,
+# 'in', iteration) so transform code can use either style interchangeably.
+_MNS_CLASS_DEF = """\
+class _MNS:
+    def __init__(self, **kw): self.__dict__.update(kw)
+    def __getitem__(self, k): return self.__dict__[k]
+    def __setitem__(self, k, v): self.__dict__[k] = v
+    def __delitem__(self, k): del self.__dict__[k]
+    def __contains__(self, k): return k in self.__dict__
+    def __iter__(self): return iter(self.__dict__)
+    def __len__(self): return len(self.__dict__)
+    def keys(self): return self.__dict__.keys()
+    def values(self): return self.__dict__.values()
+    def items(self): return self.__dict__.items()
+    def get(self, k, d=None): return self.__dict__.get(k, d)
+    def __repr__(self): return 'namespace(' + ', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()) + ')'
+"""
+
 _GET_TRANSFORMER_IMPL = """\
 def _make_get_transformer(transforms_registry, main_python, sandbox_base,
                            non_python_harness, parent_call_stack, self_key=None):
@@ -86,6 +105,7 @@ def get_transformer(bblock_id, transform_id):
                         [_sys.executable, '-m', 'pip', 'install', '--disable-pip-version-check',
                          *pip_deps],
                         check=True,
+                        capture_output=True,
                     )
 
                 def _make_python_callable(
@@ -113,7 +133,7 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
                             _tm = _types.SimpleNamespace(
                                 source_mime_type=None,
                                 target_mime_type=None,
-                                metadata=_types.SimpleNamespace(**_base_meta),
+                                metadata=_MNS(**_base_meta),
                                 context=_ctx,
                             )
                             if isinstance(content, bytes):
@@ -153,6 +173,7 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
                         _subprocess.run(
                             ['npm', 'install', '--prefix', node_dir, *npm_deps],
                             check=True,
+                            capture_output=True,
                         )
                         _sandbox_dir_str = _os.path.dirname(node_dir)
 
@@ -245,7 +266,7 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
     "import sys as _sys, json as _json, base64 as _b64, traceback as _tb,"
     " types as _types, os as _os, subprocess as _subprocess, io as _io\n"
     "from pathlib import Path as _Path\n\n"
-) + _GET_TRANSFORMER_IMPL + """
+) + _MNS_CLASS_DEF + _GET_TRANSFORMER_IMPL + """
 
 _req = _json.loads(_sys.stdin.buffer.read())
 _t_type = _req['type']
@@ -276,7 +297,7 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
     _tm = _types.SimpleNamespace(
         source_mime_type=_req.get('source_mime_type'),
         target_mime_type=_req.get('target_mime_type'),
-        metadata=_types.SimpleNamespace(**_meta_dict),
+        metadata=_MNS(**_meta_dict),
         context=_ctx_ns,
     )
     _ns = {'transform_metadata': _tm, 'input_data': _input, 'output_data': None,
@@ -352,7 +373,7 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
 _HARNESS_IMPORTS = (
     "import sys as _sys, json as _json, types as _types, base64 as _b64,"
     " io as _io, traceback as _tb, os as _os, subprocess as _subprocess\n"
-)
+) + _MNS_CLASS_DEF
 
 # Static body of the persistent harness: get_transformer factory call + the stdin request loop.
 # Written as a plain string so braces don't need escaping; dynamic values (_TRANSFORM_CODE,
@@ -377,6 +398,8 @@ def _callable(content, source_mime_type=None, extra_metadata=None):
     _d = _req['metadata']
     if isinstance(_d.get('context'), dict):
         _d['context'] = _types.SimpleNamespace(**_d['context'])
+    if isinstance(_d.get('metadata'), dict):
+        _d['metadata'] = _MNS(**_d['metadata'])
     transform_metadata = _types.SimpleNamespace(**_d)
     _raw = _b64.b64decode(_req['input'])
     try:
@@ -466,7 +489,15 @@ def _send_raw(self, req_line: bytes) -> dict | None:
             self._proc.stdin.flush()
             resp_line = self._proc.stdout.readline()
             if resp_line:
-                return json.loads(resp_line)
+                try:
+                    return json.loads(resp_line)
+                except json.JSONDecodeError as e:
+                    return {
+                        'output': None,
+                        'success': False,
+                        'stderr': f'Subprocess wrote invalid JSON ({e}): {resp_line!r}',
+                        'binary': False,
+                    }
         except (BrokenPipeError, OSError):
             pass
         return None