feat: Add strptime_to_utc and strftime functions to _singerlib.utils

singer_sdk/_singerlib/__init__.py

@@ -17,6 +17,7 @@
     write_message,
 )
 from singer_sdk._singerlib.schema import Schema, resolve_schema_references
+from singer_sdk._singerlib.utils import strftime, strptime_to_utc
 
 __all__ = [
     "Catalog",
@@ -35,4 +36,6 @@
     "write_message",
     "Schema",
     "resolve_schema_references",
+    "strftime",
+    "strptime_to_utc",
 ]

singer_sdk/_singerlib/utils.py

@@ -0,0 +1,47 @@
+from datetime import datetime, timedelta
+
+import dateutil.parser
+import pytz
+
+DATETIME_FMT = "%04Y-%m-%dT%H:%M:%S.%fZ"
+DATETIME_FMT_SAFE = "%Y-%m-%dT%H:%M:%S.%fZ"
+
+
+def strptime_to_utc(dtimestr: str) -> datetime:
+    """Parses a provide datetime string into a UTC datetime object.
+
+    Args:
+        dtimestr: a string representation of a datetime
+
+    Returns:
+        A UTC datetime.datetime object
+    """
+    d_object: datetime = dateutil.parser.parse(dtimestr)
+    if d_object.tzinfo is None:
+        return d_object.replace(tzinfo=pytz.UTC)
+    else:
+        return d_object.astimezone(tz=pytz.UTC)
+
+
+def strftime(dtime: datetime, format_str: str = DATETIME_FMT) -> str:
+    """Formats a provided datetime object as a string.
+
+    Args:
+        dtime: a datetime
+        format_str: output format specification
+
+    Returns:
+        A string in the specified format
+
+    Raises:
+        Exception: if the datetime is not UTC (if it has a nonzero time zone offset)
+    """
+    if dtime.utcoffset() != timedelta(0):
+        raise Exception("datetime must be pegged at UTC tzoneinfo")
+
+    try:
+        dt_str = dtime.strftime(format_str)
+        if dt_str.startswith("4Y"):
+            return dtime.strftime(DATETIME_FMT_SAFE)
+    except ValueError:
+        return dtime.strftime(DATETIME_FMT_SAFE)

tests/_singerlib/test_utils.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+
+import pytest
+import pytz
+
+from singer_sdk._singerlib import strftime, strptime_to_utc
+
+
+def test_small_years():
+    assert (
+        strftime(datetime(90, 1, 1, tzinfo=pytz.UTC)) == "0090-01-01T00:00:00.000000Z"
+    )
+
+
+def test_round_trip():
+    now = datetime.utcnow().replace(tzinfo=pytz.UTC)
+    dtime = strftime(now)
+    parsed_datetime = strptime_to_utc(dtime)
+    formatted_datetime = strftime(parsed_datetime)
+    assert dtime == formatted_datetime