fix: timestamp parsing for RTDB events

src/firebase_functions/db_fn.py

@@ -17,7 +17,6 @@
 
 # pylint: disable=protected-access
 import dataclasses as _dataclass
-import datetime as _dt
 import functools as _functools
 import typing as _typing
 
@@ -126,10 +125,7 @@ def _db_endpoint_handler(
         id=event_attributes["id"],
         source=event_attributes["source"],
         type=event_attributes["type"],
-        time=_dt.datetime.strptime(
-            event_attributes["time"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
+        time=_util.timestamp_conversion(event_attributes["time"]),
         data=database_event_data,
         subject=event_attributes["subject"],
         params=params,

src/firebase_functions/private/util.py

@@ -348,80 +348,33 @@ def firebase_config() -> None | FirebaseConfig:
     return FirebaseConfig(storage_bucket=json_data.get("storageBucket"))
 
 
-def nanoseconds_timestamp_conversion(time: str) -> _dt.datetime:
-    """Converts a nanosecond timestamp and returns a datetime object of the current time in UTC"""
+def normalize_timestamp_string(time: str) -> str:
+    """Truncate sub-second precision to microseconds for standard datetime parsing."""
+    # Python's %z parser accepts uppercase "Z" for UTC, but not lowercase "z".
+    if time.endswith("z"):
+        time = time[:-1] + "Z"
 
-    # Separate the date and time part from the nanoseconds.
-    datetime_str, nanosecond_str = time.replace("Z", "").replace("z", "").split(".")
-    # Parse the date and time part of the string.
-    event_time = _dt.datetime.strptime(datetime_str, "%Y-%m-%dT%H:%M:%S")
-    # Add the microseconds and timezone.
-    event_time = event_time.replace(microsecond=int(nanosecond_str[:6]), tzinfo=_dt.timezone.utc)
+    # Whole-second timestamps already fit the standard parser.
+    if "." not in time:
+        return time
 
-    return event_time
+    # Split once so the suffix still contains both fraction and timezone.
+    prefix, suffix = time.split(".", 1)
+    # The suffix must start with fractional second digits, followed by a timezone.
+    if not (fraction_match := _re.match(r"(\d+)(.*)", suffix)):
+        raise ValueError(f"Invalid timestamp format: {time}")
 
-
-def second_timestamp_conversion(time: str) -> _dt.datetime:
-    """Converts a second timestamp and returns a datetime object of the current time in UTC"""
-    return _dt.datetime.strptime(
-        time,
-        "%Y-%m-%dT%H:%M:%S%z",
-    )
-
-
-class PrecisionTimestamp(_enum.Enum):
-    """
-    The status of a token.
-    """
-
-    NANOSECONDS = "NANOSECONDS"
-
-    MICROSECONDS = "MICROSECONDS"
-
-    SECONDS = "SECONDS"
-
-    def __str__(self) -> str:
-        return self.value
-
-
-def get_precision_timestamp(time: str) -> PrecisionTimestamp:
-    """Return a bool which indicates if the timestamp is in nanoseconds"""
-    # Split the string into date-time and fraction of second
-    try:
-        _, s_fraction = time.split(".")
-    except ValueError:
-        return PrecisionTimestamp.SECONDS
-
-    # Split the fraction from the timezone specifier ('Z' or 'z')
-    s_fraction, _ = s_fraction.split("Z") if "Z" in s_fraction else s_fraction.split("z")
-
-    # If the fraction is more than 6 digits long, it's a nanosecond timestamp
-    if len(s_fraction) > 6:
-        return PrecisionTimestamp.NANOSECONDS
-    else:
-        return PrecisionTimestamp.MICROSECONDS
+    # datetime only stores microseconds, so keep the first 6 fractional digits.
+    digits, timezone = fraction_match.groups()
+    return f"{prefix}.{digits[:6]}{timezone}"
 
 
 def timestamp_conversion(time: str) -> _dt.datetime:
-    """Converts a timestamp and returns a datetime object of the current time in UTC"""
-    precision_timestamp = get_precision_timestamp(time)
-
-    if precision_timestamp == PrecisionTimestamp.NANOSECONDS:
-        return nanoseconds_timestamp_conversion(time)
-    elif precision_timestamp == PrecisionTimestamp.MICROSECONDS:
-        return microsecond_timestamp_conversion(time)
-    elif precision_timestamp == PrecisionTimestamp.SECONDS:
-        return second_timestamp_conversion(time)
-
-    raise ValueError("Invalid timestamp")
-
-
-def microsecond_timestamp_conversion(time: str) -> _dt.datetime:
-    """Converts a microsecond timestamp and returns a datetime object of the current time in UTC"""
-    return _dt.datetime.strptime(
-        time,
-        "%Y-%m-%dT%H:%M:%S.%f%z",
-    )
+    """Converts an ISO 8601 timestamp and returns a timezone-aware datetime object."""
+    normalized_time = normalize_timestamp_string(time)
+    if "." not in normalized_time:
+        return _dt.datetime.strptime(normalized_time, "%Y-%m-%dT%H:%M:%S%z")
+    return _dt.datetime.strptime(normalized_time, "%Y-%m-%dT%H:%M:%S.%f%z")
 
 
 def normalize_path(path: str) -> str:

tests/test_db.py

@@ -2,6 +2,7 @@
 Tests for the db module.
 """
 
+import datetime as dt
 import unittest
 from unittest import mock
 
@@ -81,3 +82,37 @@ def test_missing_auth_context(self):
         self.assertIsNotNone(event_arg)
         self.assertEqual(event_arg.auth_type, "unknown")
         self.assertIsNone(event_arg.auth_id)
+
+    def test_written_event_parses_timestamp_without_microseconds(self):
+        func = mock.Mock(__name__="example_func_no_microseconds")
+        decorated_func = db_fn.on_value_written(reference="/items/{itemId}")(func)
+
+        event = CloudEvent(
+            attributes={
+                "specversion": "1.0",
+                "id": "issue-257-repro",
+                "source": "//firebase.test/projects/demo-test/instances/my-instance/refs/items/123",
+                "subject": "refs/items/123",
+                "type": "google.firebase.database.ref.v1.written",
+                "time": "2025-10-30T21:15:51Z",
+                "instance": "my-instance",
+                "ref": "/items/123",
+                "firebasedatabasehost": "my-instance.firebaseio.com",
+                "location": "location",
+            },
+            data={
+                "data": {"existing": True},
+                "delta": {"updated": True},
+            },
+        )
+
+        decorated_func(event)
+
+        func.assert_called_once()
+        event_arg = func.call_args.args[0]
+        self.assertEqual(
+            event_arg.time,
+            dt.datetime(2025, 10, 30, 21, 15, 51, tzinfo=dt.timezone.utc),
+        )
+        self.assertEqual(event_arg.data.after, {"existing": True, "updated": True})
+        self.assertEqual(event_arg.params, {"itemId": "123"})

tests/test_util.py

@@ -19,15 +19,11 @@
 from os import environ, path
 
 from firebase_functions.private.util import (
-    PrecisionTimestamp,
     _unsafe_decode_id_token,
     deep_merge,
     firebase_config,
-    get_precision_timestamp,
-    microsecond_timestamp_conversion,
-    nanoseconds_timestamp_conversion,
     normalize_path,
-    second_timestamp_conversion,
+    timestamp_conversion,
 )
 
 test_bucket = "python-functions-testing.appspot.com"
@@ -56,88 +52,70 @@ def test_firebase_config_loads_from_env_file():
     )
 
 
-def test_microsecond_conversion():
+def test_timestamp_conversion_supported_formats():
     """
-    Testing microsecond_timestamp_conversion works as intended
+    Testing shared timestamp conversion handles supported RTDB and CloudEvent formats.
     """
     timestamps = [
-        ("2023-06-20T10:15:22.396358Z", "2023-06-20T10:15:22.396358Z"),
-        ("2021-02-20T11:23:45.987123Z", "2021-02-20T11:23:45.987123Z"),
-        ("2022-09-18T09:15:38.246824Z", "2022-09-18T09:15:38.246824Z"),
-        ("2010-09-18T09:15:38.246824Z", "2010-09-18T09:15:38.246824Z"),
+        (
+            "2024-04-10T12:00:00.000Z",
+            _dt.datetime(2024, 4, 10, 12, 0, tzinfo=_dt.timezone.utc),
+        ),
+        (
+            "2024-04-10T12:00:00.123456Z",
+            _dt.datetime(2024, 4, 10, 12, 0, 0, 123456, tzinfo=_dt.timezone.utc),
+        ),
+        (
+            "2024-04-10T12:00:00.123456+05:30",
+            _dt.datetime(
+                2024,
+                4,
+                10,
+                12,
+                0,
+                0,
+                123456,
+                tzinfo=_dt.timezone(_dt.timedelta(hours=5, minutes=30)),
+            ),
+        ),
+        (
+            "2024-04-10T12:00:00.123456-0700",
+            _dt.datetime(
+                2024,
+                4,
+                10,
+                12,
+                0,
+                0,
+                123456,
+                tzinfo=_dt.timezone(-_dt.timedelta(hours=7)),
+            ),
+        ),
+        (
+            "2023-01-01T12:34:56.123456789Z",
+            _dt.datetime(2023, 1, 1, 12, 34, 56, 123456, tzinfo=_dt.timezone.utc),
+        ),
+        (
+            "2023-01-01T12:34:56.123456789+05:30",
+            _dt.datetime(
+                2023,
+                1,
+                1,
+                12,
+                34,
+                56,
+                123456,
+                tzinfo=_dt.timezone(_dt.timedelta(hours=5, minutes=30)),
+            ),
+        ),
+        (
+            "2025-10-30T21:15:51Z",
+            _dt.datetime(2025, 10, 30, 21, 15, 51, tzinfo=_dt.timezone.utc),
+        ),
     ]
 
-    for input_timestamp, expected_output in timestamps:
-        expected_datetime = _dt.datetime.strptime(expected_output, "%Y-%m-%dT%H:%M:%S.%fZ")
-        expected_datetime = expected_datetime.replace(tzinfo=_dt.timezone.utc)
-        assert microsecond_timestamp_conversion(input_timestamp) == expected_datetime
-
-
-def test_nanosecond_conversion():
-    """
-    Testing nanoseconds_timestamp_conversion works as intended
-    """
-    timestamps = [
-        ("2023-01-01T12:34:56.123456789Z", "2023-01-01T12:34:56.123456Z"),
-        ("2023-02-14T14:37:52.987654321Z", "2023-02-14T14:37:52.987654Z"),
-        ("2023-03-21T06:43:58.564738291Z", "2023-03-21T06:43:58.564738Z"),
-        ("2023-08-15T22:22:22.222222222Z", "2023-08-15T22:22:22.222222Z"),
-    ]
-
-    for input_timestamp, expected_output in timestamps:
-        expected_datetime = _dt.datetime.strptime(expected_output, "%Y-%m-%dT%H:%M:%S.%fZ")
-        expected_datetime = expected_datetime.replace(tzinfo=_dt.timezone.utc)
-        assert nanoseconds_timestamp_conversion(input_timestamp) == expected_datetime
-
-
-def test_second_conversion():
-    """
-    Testing seconds_timestamp_conversion works as intended
-    """
-    timestamps = [
-        ("2023-01-01T12:34:56Z", "2023-01-01T12:34:56Z"),
-        ("2023-02-14T14:37:52Z", "2023-02-14T14:37:52Z"),
-        ("2023-03-21T06:43:58Z", "2023-03-21T06:43:58Z"),
-        ("2023-10-06T07:00:00Z", "2023-10-06T07:00:00Z"),
-    ]
-
-    for input_timestamp, expected_output in timestamps:
-        expected_datetime = _dt.datetime.strptime(expected_output, "%Y-%m-%dT%H:%M:%SZ")
-        expected_datetime = expected_datetime.replace(tzinfo=_dt.timezone.utc)
-        assert second_timestamp_conversion(input_timestamp) == expected_datetime
-
-
-def test_is_nanoseconds_timestamp():
-    """
-    Testing is_nanoseconds_timestamp works as intended
-    """
-    microsecond_timestamp1 = "2023-06-20T10:15:22.396358Z"
-    microsecond_timestamp2 = "2021-02-20T11:23:45.987123Z"
-    microsecond_timestamp3 = "2022-09-18T09:15:38.246824Z"
-    microsecond_timestamp4 = "2010-09-18T09:15:38.246824Z"
-
-    nanosecond_timestamp1 = "2023-01-01T12:34:56.123456789Z"
-    nanosecond_timestamp2 = "2023-02-14T14:37:52.987654321Z"
-    nanosecond_timestamp3 = "2023-03-21T06:43:58.564738291Z"
-    nanosecond_timestamp4 = "2023-08-15T22:22:22.222222222Z"
-
-    second_timestamp1 = "2023-01-01T12:34:56Z"
-    second_timestamp2 = "2023-02-14T14:37:52Z"
-    second_timestamp3 = "2023-03-21T06:43:58Z"
-    second_timestamp4 = "2023-08-15T22:22:22Z"
-
-    assert get_precision_timestamp(microsecond_timestamp1) is PrecisionTimestamp.MICROSECONDS
-    assert get_precision_timestamp(microsecond_timestamp2) is PrecisionTimestamp.MICROSECONDS
-    assert get_precision_timestamp(microsecond_timestamp3) is PrecisionTimestamp.MICROSECONDS
-    assert get_precision_timestamp(microsecond_timestamp4) is PrecisionTimestamp.MICROSECONDS
-    assert get_precision_timestamp(nanosecond_timestamp1) is PrecisionTimestamp.NANOSECONDS
-    assert get_precision_timestamp(nanosecond_timestamp2) is PrecisionTimestamp.NANOSECONDS
-    assert get_precision_timestamp(nanosecond_timestamp3) is PrecisionTimestamp.NANOSECONDS
-    assert get_precision_timestamp(nanosecond_timestamp4) is PrecisionTimestamp.NANOSECONDS
-    assert get_precision_timestamp(second_timestamp1) is PrecisionTimestamp.SECONDS
-    assert get_precision_timestamp(second_timestamp2) is PrecisionTimestamp.SECONDS
-    assert get_precision_timestamp(second_timestamp3) is PrecisionTimestamp.SECONDS
-    assert get_precision_timestamp(second_timestamp4) is PrecisionTimestamp.SECONDS
+    for input_timestamp, expected_datetime in timestamps:
+        assert timestamp_conversion(input_timestamp) == expected_datetime
 
 
 def test_normalize_document_path():