refactor: replace datetime parsing with utility function for consistency across event handlers

Author: CorieW

src/firebase_functions/eventarc_fn.py

@@ -14,7 +14,6 @@
 """Cloud functions to handle Eventarc events."""
 
 # pylint: disable=protected-access
-import datetime as _dt
 import functools as _functools
 import typing as _typing
 
@@ -64,10 +63,7 @@ def on_custom_event_published_wrapped(raw: _ce.CloudEvent):
                 source=event_dict["source"],
                 specversion=event_dict["specversion"],
                 subject=event_dict["subject"] if "subject" in event_dict else None,
-                time=_dt.datetime.strptime(
-                    event_dict["time"],
-                    "%Y-%m-%dT%H:%M:%S.%f%z",
-                ),
+                time=_util.timestamp_conversion(event_dict["time"]),
                 type=event_dict["type"],
             )
             _with_init(func)(event)

src/firebase_functions/pubsub_fn.py

@@ -18,7 +18,6 @@
 # pylint: disable=protected-access
 import base64 as _base64
 import dataclasses as _dataclasses
-import datetime as _dt
 import functools as _functools
 import json as _json
 import typing as _typing
@@ -105,25 +104,9 @@ def _message_handler(
     data = event_dict["data"]
     message_dict = data["message"]
 
-    # if no microseconds are present, we should set them to 0 to prevent parsing from failing
-    if "." not in event_dict["time"]:
-        event_dict["time"] = event_dict["time"].replace("Z", ".000000Z")
-    if "." not in message_dict["publish_time"]:
-        message_dict["publish_time"] = message_dict["publish_time"].replace("Z", ".000000Z")
-
-    time = _dt.datetime.strptime(
-        event_dict["time"],
-        "%Y-%m-%dT%H:%M:%S.%f%z",
-    )
-
-    publish_time = _dt.datetime.strptime(
-        message_dict["publish_time"],
-        "%Y-%m-%dT%H:%M:%S.%f%z",
-    )
-
     # Convert the UTC string into a datetime object
-    event_dict["time"] = time
-    message_dict["publish_time"] = publish_time
+    event_dict["time"] = _util.timestamp_conversion(event_dict["time"])
+    message_dict["publish_time"] = _util.timestamp_conversion(message_dict["publish_time"])
 
     # Pop unnecessary keys from the message data
     # (we get these keys from the snake case alternatives that are provided)

src/firebase_functions/remote_config_fn.py

@@ -164,7 +164,7 @@ def _config_handler(func: _C1, raw: _ce.CloudEvent) -> None:
 
     config_data = ConfigUpdateData(
         version_number=event_data["versionNumber"],
-        update_time=_dt.datetime.strptime(event_data["updateTime"], "%Y-%m-%dT%H:%M:%S.%f%z"),
+        update_time=_util.timestamp_conversion(event_data["updateTime"]),
         update_user=ConfigUser(
             name=event_data["updateUser"]["name"],
             email=event_data["updateUser"]["email"],
@@ -182,10 +182,7 @@ def _config_handler(func: _C1, raw: _ce.CloudEvent) -> None:
         source=event_dict["source"],
         specversion=event_dict["specversion"],
         subject=event_dict["subject"] if "subject" in event_dict else None,
-        time=_dt.datetime.strptime(
-            event_dict["time"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
+        time=_util.timestamp_conversion(event_dict["time"]),
         type=event_dict["type"],
     )
 

src/firebase_functions/scheduler_fn.py

@@ -105,24 +105,11 @@ def on_schedule_wrapped(request: _Request) -> _Response:
                 schedule_time = _dt.datetime.now(_timezone.utc)
             else:
                 try:
-                    # Try to parse with the stdlib which supports fractional
-                    # seconds and offsets in Python 3.11+ via fromisoformat.
-                    # Normalize RFC3339 'Z' to '+00:00' for fromisoformat.
-                    iso_str = schedule_time_str
-                    if iso_str.endswith("Z"):
-                        iso_str = iso_str[:-1] + "+00:00"
-                    schedule_time = _dt.datetime.fromisoformat(iso_str)
-                except ValueError:
-                    # Fallback to strict parsing without fractional seconds
-                    try:
-                        schedule_time = _dt.datetime.strptime(
-                            schedule_time_str,
-                            "%Y-%m-%dT%H:%M:%S%z",
-                        )
-                    except ValueError as e:
-                        # If all parsing fails, log and use current UTC time
-                        _logging.exception(e)
-                        schedule_time = _dt.datetime.now(_timezone.utc)
+                    schedule_time = _util.timestamp_conversion(schedule_time_str)
+                except ValueError as e:
+                    # If parsing fails, log and use current UTC time.
+                    _logging.exception(e)
+                    schedule_time = _dt.datetime.now(_timezone.utc)
             event = ScheduledEvent(
                 job_name=request.headers.get("X-CloudScheduler-JobName"),
                 schedule_time=schedule_time,

src/firebase_functions/test_lab_fn.py

@@ -214,7 +214,7 @@ def _event_handler(func: _C1, raw: _ce.CloudEvent) -> None:
     event_dict = {**event_data, **event_attributes}
 
     test_lab_data = TestMatrixCompletedData(
-        create_time=_dt.datetime.strptime(event_data["createTime"], "%Y-%m-%dT%H:%M:%S.%f%z"),
+        create_time=_util.timestamp_conversion(event_data["createTime"]),
         state=TestState(event_data["state"]),
         invalid_matrix_details=event_data.get("invalidMatrixDetails"),
         outcome_summary=OutcomeSummary(event_data["outcomeSummary"]),
@@ -237,10 +237,7 @@ def _event_handler(func: _C1, raw: _ce.CloudEvent) -> None:
         source=event_dict["source"],
         specversion=event_dict["specversion"],
         subject=event_dict["subject"] if "subject" in event_dict else None,
-        time=_dt.datetime.strptime(
-            event_dict["time"],
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ),
+        time=_util.timestamp_conversion(event_dict["time"]),
         type=event_dict["type"],
     )