WIP DSL

Pipfile

@@ -12,3 +12,4 @@ pytest-cov = "==3.0.0"
 freezegun = "==1.2.2"
 pyyaml = "==6.0.1"
 pre-commit = "~=2.21.0"
+black = "*"

fhirpathpy/__init__.py

@@ -3,6 +3,7 @@
 from fhirpathpy.engine import do_eval
 from fhirpathpy.engine.util import arraify, get_data, set_paths
 from fhirpathpy.engine.nodes import FP_Type
+from fhirpathpy.dsl_impl import DSL
 
 __title__ = "fhirpathpy"
 __version__ = "0.2.2"
@@ -63,6 +64,8 @@ def evaluate(resource, path, context={}, model=None):
     int: Description of return value
 
     """
+    if isinstance(path, DSL):
+        path = str(path)
     node = parse(path)
     return apply_parsed_path(resource, node, context, model)
 
@@ -82,3 +85,6 @@ def compile(path, model=None):
     For example, you could pass in the result of require("fhirpath/fhir-context/r4")
     """
     return set_paths(apply_parsed_path, parsedPath=parse(path), model=model)
+
+
+dsl = DSL()

fhirpathpy/dsl_impl.py

@@ -0,0 +1,68 @@
+class DSL:
+    path: str
+
+    def __init__(self, prefix=None):
+        self.path = prefix
+
+    def __getattr__(self, attr):
+        if self.path is None and attr == "empty":
+            return DSL("{}")
+        if self.path is None and attr == "env":
+            return DSL("%")
+        elif self.path == "%":
+            return DSL(f"{self.path}{attr}")
+        elif self.path is not None:
+            if attr == "not_":
+                return DSL(f"{self.path}.not")
+            return DSL(f"{self.path}.{attr}")
+        else:
+            return DSL(attr)
+
+    def __getitem__(self, attr):
+        return DSL(f"{self.path}[{attr}]")
+
+    def __eq__(self, other):
+        return DSL(f"{self.path} = {format_value(other)}")
+
+    def __ne__(self, other):
+        return DSL(f"{self.path} != {format_value(other)}")
+
+    def __lshift__(self, item):
+        return DSL(f"{self.path} contains {format_value(item)}")
+
+    def __or__(self, item):
+        return DSL(f"({self.path} or {format_value(item)})")
+
+    def __not__(self):
+        return DSL(f"{self.path}.not()")
+
+    def __call__(self, *args, **kwargs):
+        if len(args) == 0 and len(kwargs) == 0:
+            return DSL(f"{self.path}()")
+        elif len(args) == 0 and len(kwargs) == 1:
+            arg = build_args(kwargs)
+            return DSL(f"{self.path}({arg})")
+        elif len(args) == 1 and len(kwargs) == 0:
+            arg = format_value(args[0])
+            return DSL(f"{self.path}({arg})")
+        else:
+            raise Exception(f"Wrong arguements args={args} kwargs={kwargs}")
+
+    def __str__(self):
+        return self.path
+
+    def __add__(self, item):
+        return DSL(f"({self.path} + {format_value(item)})")
+
+    def __radd__(self, item):
+        return DSL(f"({format_value(item)} + {self.path})")
+
+
+def format_value(value):
+    if isinstance(value, str):
+        return f"'{value}'"
+    return str(value)
+
+
+def build_args(args):
+    return ",".join(f"{k}={format_value(v)}" for k, v in args.items())

tests/test_dsl.py

@@ -0,0 +1,44 @@
+from fhirpathpy import dsl
+
+
+def dsl_test():
+    assert str(dsl.Patient.id) == "Patient.id"
+    assert (
+        str(dsl.Patient.name.where(use="usual").given.first())
+        == "Patient.name.where(use='usual').given.first()"
+    )
+    assert (
+        str(
+            dsl.env.Source.entry[0].resource.expansion.contains.where(code=dsl.env.Coding.code)
+            != dsl.empty
+        )
+        == "%Source.entry[0].resource.expansion.contains.where(code=%Coding.code) != {}"
+    )
+
+    assert (
+        str(dsl.Location.where(dsl.physicalType.coding.code << "817").name)
+        == "Location.where(physicalType.coding.code contains '817').name"
+    )
+    assert (
+        str(
+            (
+                (
+                    dsl.env.QuestionnaireResponse.repeat(dsl.item)
+                    .where(linkId="test-type")
+                    .answer.children()
+                    .Coding.code
+                    == "trainingTest"
+                )
+                | (
+                    dsl.env.QuestionnaireResponse.repeat(dsl.item)
+                    .where(linkId="sample-set")
+                    .answer.count()
+                    == 6
+                )
+            ).not_()
+        )
+        == "(%QuestionnaireResponse.repeat(item).where(linkId='test-type').answer.children().Coding.code = 'trainingTest' or %QuestionnaireResponse.repeat(item).where(linkId='sample-set').answer.count() = 6).not()"
+    )
+
+    assert str(dsl.Patient.id + "foo") == "(Patient.id + 'foo')"
+    assert str("/Patient/" + dsl.Patient.id) == "('/Patient/' + Patient.id)"

tests/test_evaluators.py

@@ -6,6 +6,8 @@
 from fhirpathpy import evaluate
 from fhirpathpy.engine.invocations.constants import constants
 
+from fhirpathpy import dsl
+
 
 @pytest.mark.parametrize(
     ("resource", "path", "expected"),
@@ -212,7 +214,7 @@ def now_function_test():
         old_now_value = evaluate({}, 'now()')
         frozen_datetime.tick(1.0)
         new_now_value = evaluate({}, 'now()')
-    
+
     assert old_now_value != new_now_value
 
 
@@ -246,3 +248,14 @@ def combining_functions_test(resource, path, expected):
 )
 def path_functions_test(resource, path, expected):
     assert evaluate(resource, path) == expected
+
+
+@pytest.mark.parametrize(
+    ("resource", "path", "expected"),
+    [
+        ({"a": "lorem ipsum"}, "a.contains('sum')", [True]),
+        ({"a": "lorem ipsum"}, dsl.a.contains('sum'), [True]),
+    ],
+)
+def convert_dsl_path_test(resource, path, expected):
+    assert evaluate(resource, path) == expected