Ontology Create Link

1"""ontology-create-link — link two ontology objects with a typed edge.23Creates a ``linkTypeId`` edge from ``sourceObjectId`` to ``targetObjectId`` via4the gais.ontology SDK (``Gais.ontology.create_link``). Both object ids are5UUID-validated here (defense in depth) so ids wired from upstream nodes can't6rewrite the request route. If the server refuses to downgrade an existing7higher-confidence link, ``downgrade_refused`` is True and the id of the8pre-existing link is returned. Privileged IO carries the per-execution9USER_TOKEN inside the SDK — this node imports no HTTP client.10"""1112from __future__ import annotations1314import json15import re16import sys17import traceback18from typing import Any1920from gais import Gais2122_UUID_RE = re.compile(23    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",24    re.IGNORECASE,25)262728def _normalize_uuid(value: Any, field: str) -> str:29    """Accept a bare UUID or a ``mention://.../<uuid>`` URI; validate as UUID."""30    if not isinstance(value, str):31        raise ValueError(f"{field} must be a string, got {type(value).__name__}")32    s = value.strip()33    if "://" in s:34        s = s.rsplit("/", 1)[-1]35    if not _UUID_RE.match(s):36        raise ValueError(f"{field} is not a valid UUID: {value!r}")37    return s383940def _parse_obj(value: Any, field: str) -> dict:41    """Accept a mapping or a JSON-object string; reject anything else."""42    if value in (None, ""):43        return {}44    if isinstance(value, dict):45        return value46    if isinstance(value, str):47        parsed = json.loads(value)48        if not isinstance(parsed, dict):49            raise ValueError(f"{field} must be a JSON object")50        return parsed51    raise ValueError(f"{field} must be a JSON object")525354def _parse_confidence(value: Any) -> float | None:55    if value in (None, ""):56        return None57    return float(value)585960def process(inputs: dict[str, Any]) -> dict[str, Any]:61    source_object_id = _normalize_uuid(inputs.get("sourceObjectId"), "sourceObjectId")62    target_object_id = _normalize_uuid(inputs.get("targetObjectId"), "targetObjectId")6364    link_type_id = str(inputs.get("linkTypeId") or "").strip()65    if not link_type_id:66        raise ValueError("Required input 'linkTypeId' not provided")6768    # SDK default for properties is None (no edge props); pass None when empty.69    properties = _parse_obj(inputs.get("properties"), "properties") or None70    confidence = _parse_confidence(inputs.get("confidence"))7172    result = Gais.ontology.create_link(73        source_object_id=source_object_id,74        target_object_id=target_object_id,75        link_type_id=link_type_id,76        properties=properties,77        confidence=confidence,78    )79    if not isinstance(result, dict):80        result = {}8182    link_id = result.get("id", "")83    downgrade_refused = bool(result.get("downgrade_refused", False))8485    print(86        f"[ontology-create-link] {source_object_id} -[{link_type_id}]-> "87        f"{target_object_id} id={link_id} downgrade_refused={downgrade_refused}",88        file=sys.stderr,89    )9091    return {"id": link_id, "downgrade_refused": downgrade_refused}929394def main() -> None:95    try:96        envelope = json.loads(sys.stdin.read() or "{}")97        if not isinstance(envelope, dict):98            envelope = {}99        inputs = envelope.get("inputs", {}) or {}100        json.dump(process(inputs), sys.stdout)101    except Exception as e:102        print(103            json.dumps({104                "error": str(e),105                "errorType": type(e).__name__,106                "traceback": traceback.format_exc(),107            }),108            file=sys.stderr,109        )110        sys.exit(1)111112113if __name__ == "__main__":114    main()