Merge pull request #44 from Tiro-health/master

Support `collection.abc.Mapping` as resource instead of only `dict`

fhirpathpy/engine/evaluators/__init__.py

@@ -1,3 +1,4 @@
+from collections import abc
 from decimal import Decimal
 from functools import reduce
 
@@ -185,7 +186,7 @@ def func(acc, res):
         if isinstance(res.data, nodes.FP_Quantity):
             toAdd = res.data.value
 
-        if actualTypes and isinstance(res.data, dict):
+        if actualTypes and isinstance(res.data, abc.Mapping):
             # Use actualTypes to find the field's value
             for actualType in actualTypes:
                 field = f"{key}{actualType}"
@@ -194,7 +195,7 @@ def func(acc, res):
                 if toAdd is not None or toAdd_ is not None:
                     childPath += actualType
                     break
-        elif isinstance(res.data, dict):
+        elif isinstance(res.data, abc.Mapping):
             toAdd = res.data.get(key)
             toAdd_ = res.data.get(f"_{key}")
             if key == "extension":

fhirpathpy/engine/invocations/equality.py

@@ -1,3 +1,4 @@
+from collections import abc
 from decimal import Decimal
 import json
 import fhirpathpy.engine.util as util
@@ -80,9 +81,9 @@ def equivalence(ctx, x, y):
     if isinstance(x_val, nodes.FP_Quantity) and isinstance(y_val, nodes.FP_Quantity):
         return x_val.deep_equal(y_val)
 
-    if isinstance(a, (dict, list)) and isinstance(b, (dict, list)):
+    if isinstance(a, (abc.Mapping, list)) and isinstance(b, (abc.Mapping, list)):
         def deep_equal(a, b):
-            if isinstance(a, dict) and isinstance(b, dict):
+            if isinstance(a, abc.Mapping) and isinstance(b, abc.Mapping):
                 if a.keys() != b.keys():
                     return False
                 return all(deep_equal(a[key], b[key]) for key in a)

fhirpathpy/engine/invocations/filtering.py

@@ -1,3 +1,4 @@
+from collections import abc
 from decimal import Decimal
 import numbers
 import fhirpathpy.engine.util as util
@@ -121,7 +122,7 @@ def extension(ctx, data, url):
     res = []
     for d in data:
         element = util.get_data(d)
-        if isinstance(element, dict):
+        if isinstance(element, abc.Mapping):
             exts = [e for e in element.get("extension", []) if e["url"] == url]
             if len(exts) > 0:
                 res.append(nodes.ResourceNode.create_node(exts[0], "Extension"))

fhirpathpy/engine/invocations/navigation.py

@@ -1,3 +1,4 @@
+from collections import abc
 from functools import reduce
 import fhirpathpy.engine.util as util
 import fhirpathpy.engine.nodes as nodes
@@ -15,7 +16,7 @@ def func(acc, res):
         if isinstance(data, list):
             data = dict((i, data[i]) for i in range(0, len(data)))
 
-        if isinstance(data, dict):
+        if isinstance(data, abc.Mapping):
             for prop in data.keys():
                 value = data[prop]
                 childPath = ""

fhirpathpy/engine/nodes.py

@@ -1,3 +1,4 @@
+from collections import abc
 import copy
 from datetime import datetime, timedelta, timezone
 from dateutil.relativedelta import relativedelta
@@ -824,7 +825,7 @@ def __init__(self, data, path, _data=None):
         If data is a resource (maybe a contained resource) reset the path
         information to the resource type.
         """
-        if isinstance(data, dict) and "resourceType" in data:
+        if isinstance(data, abc.Mapping) and "resourceType" in data:
             path = data["resourceType"]
 
         self.path = path
@@ -872,7 +873,7 @@ def convert_data(self):
         cls = TypeInfo.type_to_class_with_check_string.get(self.path)
         if cls:
             data = FP_TimeBase.check_string(cls, data) or data
-        if isinstance(data, dict) and data["system"] == "http://unitsofmeasure.org":
+        if isinstance(data, abc.Mapping) and data["system"] == "http://unitsofmeasure.org":
             data = FP_Quantity(
                 data["value"],
                 FP_Quantity.timeUnitsToUCUM.get(data["code"], "'" + data["code"] + "'"),
@@ -934,7 +935,7 @@ def create_by_value_in_namespace(namespace, value):
             name = "Quantity"
         elif isinstance(value, str):
             name = "string"
-        elif isinstance(value, dict):
+        elif isinstance(value, abc.Mapping):
             name = "object"
 
         if name == "bool":

tests/test_evaluators.py

@@ -1,7 +1,10 @@
 import json
+from collections.abc import Mapping
+from dataclasses import dataclass, fields
 from datetime import datetime, timezone
-from unittest import mock
 from decimal import Decimal
+from unittest import mock
+
 import pytest
 
 from fhirpathpy import evaluate
@@ -275,3 +278,118 @@ def combining_functions_test(resource, path, expected):
 )
 def path_functions_test(resource, path, expected):
     assert evaluate(resource, path) == expected
+
+
+
+@dataclass(eq=True)
+class NestedMapping(Mapping):
+    d: int
+    e: str
+
+    def __iter__(self):
+        return iter([field.name for field in fields(self)])
+
+    def __getitem__(self, key: str):
+        try:
+            return getattr(self, key)
+        except AttributeError:
+            raise KeyError(key)
+
+    def __len__(self):
+        return len(fields(self))
+
+
+@dataclass(eq=True)
+class CustomMapping(Mapping):
+    a: int
+    b: dict
+    c: NestedMapping
+    d: tuple[NestedMapping, ...]
+    e: list[NestedMapping]
+
+    def __iter__(self):
+        return iter([field.name for field in fields(self)])
+
+    def __getitem__(self, key: str):
+        try:
+            return getattr(self, key)
+        except AttributeError:
+            raise KeyError(key)
+
+    def __len__(self):
+        return len(fields(self))
+
+
+@pytest.mark.parametrize(
+    ("resource", "path", "expected"),
+    [
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "a",
+            [1],
+        ),
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "b.c",
+            [True],
+        ),
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "c",
+            [NestedMapping(d=3, e="f")],
+        ),
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "c.d",
+            [3],
+        ),
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "e.last()",
+            [NestedMapping(d=5, e="w")],
+        ),
+        (
+            CustomMapping(
+                a=1,
+                b={"c": True},
+                c=NestedMapping(d=3, e="f"),
+                d=(NestedMapping(d=4, e="x"), NestedMapping(d=3, e="y")),
+                e=[NestedMapping(d=4, e="z"), NestedMapping(d=5, e="w")],
+            ),
+            "e.where(d=5)",
+            [NestedMapping(d=5, e="w")],
+        )
+    ],
+)
+def mappings_test(resource, path, expected):
+    assert evaluate(resource, path) == expected