Add metadata support to HarmonizationRule

src/harmonization_framework/harmonization_rule.py

@@ -1,4 +1,4 @@
-from typing import Any, List
+from typing import Any, Dict, List, Optional
 from .element import DataElement
 from .primitives.base import PrimitiveOperation
 from .primitives import PrimitiveVocabulary, Bin, Cast, ConvertDate, ConvertUnits, DoNothing, EnumToEnum, FormatNumber, NormalizeBoolean, NormalizeText, Offset, Reduce, Round, Scale, Substitute, Threshold, Truncate
@@ -8,10 +8,17 @@
 # Backwards-compatible alias
 
 class HarmonizationRule:
-    def __init__(self, source: DataElement, target: DataElement, transformation: List[PrimitiveOperation] = None):
+    def __init__(
+        self,
+        source: DataElement,
+        target: DataElement,
+        transformation: List[PrimitiveOperation] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ):
         self.source = source
         self.target = target
         self._transform = transformation
+        self.metadata = metadata
         self.serialization = json.dumps(self.serialize())
 
     def serialize(self):
@@ -22,6 +29,8 @@ def serialize(self):
                 primitive.to_dict() for primitive in (self._transform or [])
             ],
         }
+        if self.metadata:
+            output["metadata"] = self.metadata
         return output
 
     def __str__(self):
@@ -46,6 +55,7 @@ def from_serialization(cls, serialization):
         source = serialization["source"]
         target = serialization["target"]
         operations = serialization["operations"]
+        metadata = serialization.get("metadata")
         transformation = []
         for operation in operations:
             match operation["operation"]:
@@ -84,7 +94,7 @@ def from_serialization(cls, serialization):
                 case _:
                     raise ValueError(f"Unknown operation: {operation['operation']}")
             transformation.append(primitive)
-        return HarmonizationRule(source, target, transformation)
+        return HarmonizationRule(source, target, transformation, metadata=metadata)
 
 
 # Backwards-compatible alias

tests/conftest.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SRC = ROOT / "src"
+
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))

tests/test_rule_serialization.py

@@ -24,6 +24,7 @@ def test_rule_serialization_roundtrip_and_transform():
         "age_text",
         "age_years",
         [Cast("text", "integer"), Round(0)],
+        metadata={"comments": ["source: demo"], "provenance": "demo"},
     )
 
     payload = rule.serialize()