add authcontxt params to database_event (#270)

Co-authored-by: Kushal Palesha <3357451+kushalpalesha@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: kushalpalesha

src/firebase_functions/db_fn.py

@@ -34,6 +34,8 @@
 _event_type_updated = "google.firebase.database.ref.v1.updated"
 _event_type_deleted = "google.firebase.database.ref.v1.deleted"
 
+AuthType = _typing.Literal["app_user", "admin", "unauthenticated", "unknown"]
+
 
 @_dataclass.dataclass(frozen=True)
 class Event(_core.CloudEvent[T]):
@@ -67,6 +69,16 @@ class Event(_core.CloudEvent[T]):
     Only named capture groups are populated - {key}, {key=*}, {key=**}
     """
 
+    auth_type: AuthType
+    """
+    The type of principal that triggered the event.
+    """
+
+    auth_id: str | None
+    """
+    The unique identifier for the principal.
+    """
+
 
 _E1 = Event[Change[_typing.Any | None]]
 _E2 = Event[_typing.Any | None]
@@ -104,6 +116,7 @@ def _db_endpoint_handler(
         **ref_pattern.extract_matches(event_ref),
         **instance_pattern.extract_matches(event_instance),
     }
+
     database_event = Event(
         firebase_database_host=event_attributes["firebasedatabasehost"],
         instance=event_instance,
@@ -120,6 +133,8 @@ def _db_endpoint_handler(
         data=database_event_data,
         subject=event_attributes["subject"],
         params=params,
+        auth_type=event_attributes.get("authtype", "unknown"),
+        auth_id=event_attributes.get("authid"),
     )
     _core._with_init(func)(database_event)
 

tests/test_db.py

@@ -38,10 +38,46 @@ def init():
                 "ref": "ref",
                 "firebasedatabasehost": "firebasedatabasehost",
                 "location": "location",
+                "authtype": "app_user",
+                "authid": "auth-id",
             },
             data={"delta": "delta"},
         )
 
         decorated_func(event)
 
+        func.assert_called_once()
+        event_arg = func.call_args.args[0]
+        self.assertIsNotNone(event_arg)
+        self.assertEqual(event_arg.auth_type, "app_user")
+        self.assertEqual(event_arg.auth_id, "auth-id")
+
         self.assertEqual(hello, "world")
+
+    def test_missing_auth_context(self):
+        func = mock.Mock(__name__="example_func_no_auth")
+        decorated_func = db_fn.on_value_created(reference="path")(func)
+
+        event = CloudEvent(
+            attributes={
+                "specversion": "1.0",
+                "id": "id",
+                "source": "source",
+                "subject": "subject",
+                "type": "type",
+                "time": "2024-04-10T12:00:00.000Z",
+                "instance": "instance",
+                "ref": "ref",
+                "firebasedatabasehost": "firebasedatabasehost",
+                "location": "location",
+            },
+            data={"delta": "delta"},
+        )
+
+        decorated_func(event)
+
+        func.assert_called_once()
+        event_arg = func.call_args.args[0]
+        self.assertIsNotNone(event_arg)
+        self.assertEqual(event_arg.auth_type, "unknown")
+        self.assertIsNone(event_arg.auth_id)