fix: simplify timestamp parsing

Author: CorieW

src/firebase_functions/private/util.py

@@ -348,79 +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)
-
-    return event_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):
-    """
-    Timestamp precision levels supported by Firebase event timestamp parsing.
-    """
-
-    NANOSECONDS = "NANOSECONDS"
-
-    MICROSECONDS = "MICROSECONDS"
-
-    SECONDS = "SECONDS"
-
-    def __str__(self) -> str:
-        return self.value
-
-
-def get_precision_timestamp(time: str) -> PrecisionTimestamp:
-    """Return the precision used by a Firebase event timestamp."""
+    # Whole-second timestamps already fit the standard parser.
     if "." not in time:
-        return PrecisionTimestamp.SECONDS
+        return time
 
-    _, s_fraction = time.split(".", 1)
-    if not (fraction_match := _re.match(r"\d+", s_fraction)):
+    # 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}")
-    s_fraction = fraction_match.group()
 
-    # 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_util.py

@@ -19,15 +19,10 @@
 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,
 )
 
@@ -57,57 +52,6 @@ def test_firebase_config_loads_from_env_file():
     )
 
 
-def test_microsecond_conversion():
-    """
-    Testing microsecond_timestamp_conversion works as intended
-    """
-    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"),
-    ]
-
-    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_timestamp_conversion_supported_formats():
     """
     Testing shared timestamp conversion handles supported RTDB and CloudEvent formats.
@@ -151,6 +95,19 @@ def test_timestamp_conversion_supported_formats():
             "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),
@@ -161,39 +118,6 @@ def test_timestamp_conversion_supported_formats():
         assert 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
-
-
 def test_normalize_document_path():
     """
     Testing "document" path passed to Firestore event listener