Add support for user defined functions #49

Author: ruscoder

CHANGELOG.md

@@ -1,6 +1,7 @@
-## latest
+## 2.0.0
 
 - Raise an error in case of accessing undefined env variable #50 @ruscoder
+- Add support for user defined functions #49 @ruscoder
 
 ## 1.2.1
 

README.md

@@ -68,6 +68,10 @@ context (dict): a hash of variable name/value pairs.
 
 model (dict): The "model" data object specific to a domain, e.g. R4.
 
+options (dict) - Custom options (see the documentation below)
+
+options.userInvocationTable - a user invocation table used to replace any existing functions or define new ones (see User-defined functions documentation below)
+
 ## compile
 Returns a function that takes a resource and an optional context hash (see "evaluate"), and returns the result of evaluating the given FHIRPath expression on that resource.  The advantage of this function over "evaluate" is that if you have multiple resources, the given FHIRPath expression will only be parsed once
 
@@ -76,3 +80,28 @@ Returns a function that takes a resource and an optional context hash (see "eval
 path (string) - the FHIRPath expression to be parsed.
 
 model (dict) - The "model" data object specific to a domain, e.g. R4.
+
+options (dict) - Custom options
+
+options.userInvocationTable - a user invocation table used to replace any existing functions or define new ones (see User-defined functions documentation below)
+
+## User-defined functions
+
+```python
+user_invocation_table = {
+    "pow": {
+        "fn": lambda inputs, exp=2: [i**exp for i in inputs],
+        "arity": {0: [], 1: ["Integer"]},
+    }
+}
+
+result = evaluate(
+    {"a": [5, 6, 7]},
+    "a.pow()",
+    options={"userInvocationTable": user_invocation_table},
+)
+
+# result: [25, 36, 49]
+```
+
+It works similarly to [fhirpath.js](https://github.com/HL7/fhirpath.js/tree/master?tab=readme-ov-file#user-defined-functions)

fhirpathpy/__init__.py

@@ -1,11 +1,11 @@
 from fhirpathpy.engine.invocations.constants import constants
 from fhirpathpy.parser import parse
 from fhirpathpy.engine import do_eval
-from fhirpathpy.engine.util import arraify, get_data, set_paths
+from fhirpathpy.engine.util import arraify, get_data, set_paths, process_user_invocation_table
 from fhirpathpy.engine.nodes import FP_Type, ResourceNode
 
 __title__ = "fhirpathpy"
-__version__ = "1.2.1"
+__version__ = "2.0.0"
 __author__ = "beda.software"
 __license__ = "MIT"
 __copyright__ = "Copyright 2025 beda.software"
@@ -14,7 +14,7 @@
 VERSION = __version__
 
 
-def apply_parsed_path(resource, parsedPath, context=None, model=None):
+def apply_parsed_path(resource, parsedPath, context=None, model=None, options=None):
     constants.reset()
     dataRoot = arraify(resource)
 
@@ -27,7 +27,14 @@ def apply_parsed_path(resource, parsedPath, context=None, model=None):
     vars = {"context": resource, "ucum": "http://unitsofmeasure.org"}
     vars.update(context or {})
 
-    ctx = {"dataRoot": dataRoot, "vars": vars, "model": model}
+    ctx = {
+        "dataRoot": dataRoot,
+        "vars": vars,
+        "model": model,
+        "userInvocationTable": process_user_invocation_table(
+            (options or {}).get("userInvocationTable", {})
+        ),
+    }
     node = do_eval(ctx, dataRoot, parsedPath["children"][0])
 
     # Resolve any internal "ResourceNode" instances.  Continue to let FP_Type
@@ -57,7 +64,7 @@ def visit(node):
     return visit(node)
 
 
-def evaluate(resource, path, context=None, model=None):
+def evaluate(resource, path, context=None, model=None, options=None):
     """
     Evaluates the "path" FHIRPath expression on the given resource, using data
     from "context" for variables mentioned in the "path" expression.
@@ -75,14 +82,14 @@ def evaluate(resource, path, context=None, model=None):
     if isinstance(path, dict):
         node = parse(path["expression"])
         if "base" in path:
-            resource = ResourceNode.create_node(resource, path['base'])
+            resource = ResourceNode.create_node(resource, path["base"])
     else:
         node = parse(path)
 
-    return apply_parsed_path(resource, node, context or {}, model)
+    return apply_parsed_path(resource, node, context or {}, model, options)
 
 
-def compile(path, model=None):
+def compile(path, model=None, options=None):
     """
     Returns a function that takes a resource and an optional context hash (see
     "evaluate"), and returns the result of evaluating the given FHIRPath
@@ -96,4 +103,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)
+    return set_paths(
+        apply_parsed_path, parsedPath=parse(path), model=model, options=options
+    )

fhirpathpy/engine/__init__.py

@@ -3,7 +3,9 @@
 import fhirpathpy.engine.util as util
 from fhirpathpy.engine.nodes import TypeInfo
 from fhirpathpy.engine.evaluators import evaluators
-from fhirpathpy.engine.invocations import invocation_registry
+from fhirpathpy.engine.invocations import (
+    invocation_registry as base_invocation_registry,
+)
 
 
 def check_integer_param(val):
@@ -45,6 +47,11 @@ def do_eval(ctx, parentData, node):
 
 
 def doInvoke(ctx, fn_name, data, raw_params):
+    invocation_registry = {
+        **base_invocation_registry,
+        **(ctx["userInvocationTable"] or {}),
+    }
+
     if isinstance(fn_name, list) and len(fn_name) == 1:
         fn_name = fn_name[0]
 
@@ -170,7 +177,11 @@ def func(data):
 
 
 def infix_invoke(ctx, fn_name, data, raw_params):
-    if not fn_name in invocation_registry or not "fn" in invocation_registry[fn_name]:
+    invocation_registry = {
+        **base_invocation_registry,
+        **(ctx["userInvocationTable"] or {}),
+    }
+    if fn_name not in invocation_registry or "fn" not in invocation_registry[fn_name]:
         raise Exception("Not implemented " + fn_name)
 
     invocation = invocation_registry[fn_name]

fhirpathpy/engine/util.py

@@ -6,13 +6,16 @@
 
 
 class set_paths:
-    def __init__(self, func, parsedPath, model=None):
+    def __init__(self, func, parsedPath, model=None, options=None):
         self.func = func
         self.parsedPath = parsedPath
         self.model = model
+        self.options = options
 
     def __call__(self, resource, context=None):
-        return self.func(resource, self.parsedPath, context or {}, self.model)
+        return self.func(
+            resource, self.parsedPath, context or {}, self.model, self.options
+        )
 
 
 def get_data(value):
@@ -92,8 +95,21 @@ def uniq(arr):
         ordered_dict[key] = x
     return list(ordered_dict.values())
 
+
 def val_data_converted(val):
     if isinstance(val, ResourceNode):
         val = val.convert_data()
 
     return val
+
+
+def process_user_invocation_table(table):
+    return {
+        name: {
+            **entity,
+            "fn": lambda ctx, inputs, *args: entity["fn"](
+                [get_data(i) for i in inputs], *args
+            ),
+        }
+        for name, entity in table.items()
+    }

tests/test_evaluators.py

@@ -401,3 +401,26 @@ def external_constant_test():
 def external_constant_fails_on_undefined_var_test():
     with pytest.raises(ValueError):
         evaluate({}, "%var")
+
+
+def user_invocation_table_test():
+    user_invocation_table = {
+        "pow": {
+            "fn": lambda inputs, exp=2: [i ** exp for i in inputs],
+            "arity": {0: [], 1: ["Integer"]},
+        }
+    }
+
+    result = evaluate(
+        {"a": [5, 6, 7]},
+        "a.pow()",
+        options={"userInvocationTable": user_invocation_table},
+    )
+    assert result == [5 * 5, 6 * 6, 7 * 7]
+
+    result = evaluate(
+        {"a": [5, 6, 7]},
+        "a.pow(3)",
+        options={"userInvocationTable": user_invocation_table},
+    )
+    assert result == [5 * 5 * 5, 6 * 6 * 6, 7 * 7 * 7]

tests/test_real.py

@@ -174,3 +174,18 @@ def reference_filter_test():
         "Encounter.participant.individual.reference.where($this.matches('Practitioner/'))",
     )
     assert result == ["Practitioner/dr-johns"]
+
+
+def compile_with_user_defined_table_test():
+    user_invocation_table = {
+        "pow": {
+            "fn": lambda inputs, exp=2: [i ** exp for i in inputs],
+            "arity": {0: [], 1: ["Integer"]},
+        }
+    }
+
+    expr = compile(
+        "a.pow()",
+        options={"userInvocationTable": user_invocation_table},
+    )
+    assert expr({"a": [5, 6, 7]}) == [5 * 5, 6 * 6, 7 * 7]