Merge pull request #45 from invertase/main

feat(db): support path patterns

Author: taeold

src/firebase_functions/db_fn.py

@@ -20,6 +20,7 @@
 import typing as _typing
 import datetime as _dt
 import firebase_functions.private.util as _util
+import firebase_functions.private.path_pattern as _path_pattern
 import firebase_functions.core as _core
 import cloudevents.http as _ce
 
@@ -91,12 +92,12 @@ class Event(_core.CloudEvent[_core.T]):
 def _db_endpoint_handler(
     func: _C1 | _C2,
     event_type: str,
+    ref_pattern: _path_pattern.PathPattern,
+    instance_pattern: _path_pattern.PathPattern,
     raw: _ce.CloudEvent,
 ) -> None:
     event_attributes = raw._get_attributes()
     event_data: _typing.Any = raw.get_data()
-    # TODO Params are built locally via path pattern which is currently unimplemented
-    params: dict[str, str] = {}
     database_event_data = event_data
     if event_type == _event_type_deleted:
         database_event_data = database_event_data["data"]
@@ -112,10 +113,16 @@ def _db_endpoint_handler(
             before=before,
             after=after,
         )
+    event_instance = event_attributes["instance"]
+    event_ref = event_attributes["ref"]
+    params: dict[str, str] = {
+        **ref_pattern.extract_matches(event_ref),
+        **instance_pattern.extract_matches(event_instance),
+    }
     database_event = Event(
         firebase_database_host=event_attributes["firebasedatabasehost"],
-        instance=event_attributes["instance"],
-        reference=event_attributes["ref"],
+        instance=event_instance,
+        reference=event_ref,
         location=event_attributes["location"],
         specversion=event_attributes["specversion"],
         id=event_attributes["id"],
@@ -155,15 +162,27 @@ def example(event: Event[Change[object]]) -> None:
     options = DatabaseOptions(**kwargs)
 
     def on_value_written_inner_decorator(func: _C1):
+        ref_pattern = _path_pattern.PathPattern(options.reference)
+        instance_pattern = _path_pattern.PathPattern(
+            options.instance if options.instance is not None else "*")
 
         @_functools.wraps(func)
         def on_value_written_wrapped(raw: _ce.CloudEvent):
-            return _db_endpoint_handler(func, _event_type_written, raw)
+            return _db_endpoint_handler(
+                func,
+                _event_type_written,
+                ref_pattern,
+                instance_pattern,
+                raw,
+            )
 
         _util.set_func_endpoint_attr(
             on_value_written_wrapped,
-            options._endpoint(event_type=_event_type_written,
-                              func_name=func.__name__),
+            options._endpoint(
+                event_type=_event_type_written,
+                func_name=func.__name__,
+                instance_pattern=instance_pattern,
+            ),
         )
         return on_value_written_wrapped
 
@@ -193,15 +212,27 @@ def example(event: Event[Change[object]]) -> None:
     options = DatabaseOptions(**kwargs)
 
     def on_value_updated_inner_decorator(func: _C1):
+        ref_pattern = _path_pattern.PathPattern(options.reference)
+        instance_pattern = _path_pattern.PathPattern(
+            options.instance if options.instance is not None else "*")
 
         @_functools.wraps(func)
         def on_value_updated_wrapped(raw: _ce.CloudEvent):
-            return _db_endpoint_handler(func, _event_type_updated, raw)
+            return _db_endpoint_handler(
+                func,
+                _event_type_updated,
+                ref_pattern,
+                instance_pattern,
+                raw,
+            )
 
         _util.set_func_endpoint_attr(
             on_value_updated_wrapped,
-            options._endpoint(event_type=_event_type_updated,
-                              func_name=func.__name__),
+            options._endpoint(
+                event_type=_event_type_updated,
+                func_name=func.__name__,
+                instance_pattern=instance_pattern,
+            ),
         )
         return on_value_updated_wrapped
 
@@ -231,15 +262,27 @@ def example(event: Event[object]):
     options = DatabaseOptions(**kwargs)
 
     def on_value_created_inner_decorator(func: _C2):
+        ref_pattern = _path_pattern.PathPattern(options.reference)
+        instance_pattern = _path_pattern.PathPattern(
+            options.instance if options.instance is not None else "*")
 
         @_functools.wraps(func)
         def on_value_created_wrapped(raw: _ce.CloudEvent):
-            return _db_endpoint_handler(func, _event_type_created, raw)
+            return _db_endpoint_handler(
+                func,
+                _event_type_created,
+                ref_pattern,
+                instance_pattern,
+                raw,
+            )
 
         _util.set_func_endpoint_attr(
             on_value_created_wrapped,
-            options._endpoint(event_type=_event_type_created,
-                              func_name=func.__name__),
+            options._endpoint(
+                event_type=_event_type_created,
+                func_name=func.__name__,
+                instance_pattern=instance_pattern,
+            ),
         )
         return on_value_created_wrapped
 
@@ -269,15 +312,27 @@ def example(event: Event[object]) -> None:
     options = DatabaseOptions(**kwargs)
 
     def on_value_deleted_inner_decorator(func: _C2):
+        ref_pattern = _path_pattern.PathPattern(options.reference)
+        instance_pattern = _path_pattern.PathPattern(
+            options.instance if options.instance is not None else "*")
 
         @_functools.wraps(func)
         def on_value_deleted_wrapped(raw: _ce.CloudEvent):
-            return _db_endpoint_handler(func, _event_type_deleted, raw)
+            return _db_endpoint_handler(
+                func,
+                _event_type_deleted,
+                ref_pattern,
+                instance_pattern,
+                raw,
+            )
 
         _util.set_func_endpoint_attr(
             on_value_deleted_wrapped,
-            options._endpoint(event_type=_event_type_deleted,
-                              func_name=func.__name__),
+            options._endpoint(
+                event_type=_event_type_deleted,
+                func_name=func.__name__,
+                instance_pattern=instance_pattern,
+            ),
         )
         return on_value_deleted_wrapped
 

src/firebase_functions/options.py

@@ -23,6 +23,7 @@
 
 import firebase_functions.private.manifest as _manifest
 import firebase_functions.private.util as _util
+import firebase_functions.private.path_pattern as _path_pattern
 from firebase_functions.params import SecretParam, Expression
 
 USE_DEFAULT = _util.Sentinel(
@@ -394,13 +395,15 @@ def _endpoint(
         **kwargs,
     ) -> _manifest.ManifestEndpoint:
         assert kwargs["event_type"] is not None
-        event_filter_instance = self.instance if self.instance is not None else "*"
+        assert kwargs["instance_pattern"] is not None
+        instance_pattern: _path_pattern.PathPattern = kwargs["instance_pattern"]
+        event_filter_instance = instance_pattern.value
         event_filters: _typing.Any = {}
         event_filters_path_patterns: _typing.Any = {
             # Note: Eventarc always treats ref as a path pattern
             "ref": self.reference.strip("/"),
         }
-        if "*" in event_filter_instance:
+        if instance_pattern.has_wildcards:
             event_filters_path_patterns["instance"] = event_filter_instance
         else:
             event_filters["instance"] = event_filter_instance

src/firebase_functions/private/path_pattern.py

@@ -0,0 +1,182 @@
+# Copyright 2023 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Path pattern matching utilities."""
+
+from enum import Enum
+import re
+
+
+def path_parts(path: str) -> list[str]:
+    if not path or path == "" or path == "/":
+        return []
+    return path.strip("/").split("/")
+
+
+def join_path(base: str, child: str) -> str:
+    return "/".join(path_parts(base) + path_parts(child))
+
+
+def trim_param(param: str) -> str:
+    param_no_braces = param[1:-1]
+    if "=" in param_no_braces:
+        return param_no_braces[:param_no_braces.index("=")]
+    return param_no_braces
+
+
+_WILDCARD_CAPTURE_REGEX = re.compile(r"{[^/{}]+}", re.IGNORECASE)
+
+
+class SegmentName(Enum):
+    SEGMENT = "segment"
+    SINGLE_CAPTURE = "single-capture"
+    MULTI_CAPTURE = "multi-capture"
+
+
+class PathSegment:
+    """
+    A segment of a path pattern.
+    """
+    name: SegmentName
+    value: str
+    trimmed: str
+
+    def __str__(self):
+        return self.value
+
+    @property
+    def is_single_segment_wildcard(self):
+        pass
+
+    @property
+    def is_multi_segment_wildcard(self):
+        pass
+
+
+class Segment(PathSegment):
+    """
+    A segment of a path pattern.
+    """
+
+    def __init__(self, value: str):
+        self.value = value
+        self.trimmed = value
+        self.name = SegmentName.SEGMENT
+
+    @property
+    def is_single_segment_wildcard(self):
+        return "*" in self.value and not self.is_multi_segment_wildcard
+
+    @property
+    def is_multi_segment_wildcard(self):
+        return "**" in self.value
+
+
+class SingleCaptureSegment(PathSegment):
+    """
+    A segment of a path pattern that captures a single segment.
+    """
+    name = SegmentName.SINGLE_CAPTURE
+
+    def __init__(self, value):
+        self.value = value
+        self.trimmed = trim_param(value)
+
+    @property
+    def is_single_segment_wildcard(self):
+        return True
+
+    @property
+    def is_multi_segment_wildcard(self):
+        return False
+
+
+class MultiCaptureSegment(PathSegment):
+    """
+    A segment of a path pattern that captures multiple segments.
+    """
+
+    name = SegmentName.MULTI_CAPTURE
+
+    def __init__(self, value):
+        self.value = value
+        self.trimmed = trim_param(value)
+
+    @property
+    def is_single_segment_wildcard(self):
+        return False
+
+    @property
+    def is_multi_segment_wildcard(self):
+        return True
+
+
+class PathPattern:
+    """
+    Implements Eventarc's path pattern from the spec
+    https://cloud.google.com/eventarc/docs/path-patterns
+    """
+    segments: list[PathSegment]
+
+    def __init__(self, raw_path: str):
+        self.raw = raw_path
+        self.segments = []
+        self.init_path_segments(raw_path)
+
+    def init_path_segments(self, raw: str):
+        parts = raw.split("/")
+        for part in parts:
+            segment: PathSegment | None = None
+            capture = re.findall(_WILDCARD_CAPTURE_REGEX, part)
+            if capture is not None and len(capture) == 1:
+                if "**" in part:
+                    segment = MultiCaptureSegment(part)
+                else:
+                    segment = SingleCaptureSegment(part)
+            else:
+                segment = Segment(part)
+            self.segments.append(segment)
+
+    @property
+    def value(self) -> str:
+        return self.raw
+
+    @property
+    def has_wildcards(self) -> bool:
+        return any(segment.is_single_segment_wildcard or
+                   segment.is_multi_segment_wildcard
+                   for segment in self.segments)
+
+    @property
+    def has_captures(self) -> bool:
+        return any(segment.name in (SegmentName.SINGLE_CAPTURE,
+                                    SegmentName.MULTI_CAPTURE)
+                   for segment in self.segments)
+
+    def extract_matches(self, path: str) -> dict[str, str]:
+        matches: dict[str, str] = {}
+        if not self.has_captures:
+            return matches
+        path_segments = path_parts(path)
+        path_ndx = 0
+        for segment_ndx in range(len(self.segments)):
+            segment = self.segments[segment_ndx]
+            remaining_segments = len(self.segments) - 1 - segment_ndx
+            next_path_ndx = len(path_segments) - remaining_segments
+            if segment.name == SegmentName.SINGLE_CAPTURE:
+                matches[segment.trimmed] = path_segments[path_ndx]
+            elif segment.name == SegmentName.MULTI_CAPTURE:
+                matches[segment.trimmed] = "/".join(
+                    path_segments[path_ndx:next_path_ndx])
+            path_ndx = next_path_ndx if segment.is_multi_segment_wildcard else path_ndx + 1
+        return matches