diff --git a/src/firebase_functions/db_fn.py b/src/firebase_functions/db_fn.py index 4d0bbbd5..b6952f3f 100644 --- a/src/firebase_functions/db_fn.py +++ b/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, diff --git a/src/firebase_functions/private/util.py b/src/firebase_functions/private/util.py index 9df09035..d91d13e2 100644 --- a/src/firebase_functions/private/util.py +++ b/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: diff --git a/tests/test_db.py b/tests/test_db.py index 0f087cd9..94afad52 100644 --- a/tests/test_db.py +++ b/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"}) diff --git a/tests/test_util.py b/tests/test_util.py index 34d975d2..71ce6141 100644 --- a/tests/test_util.py +++ b/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():