Fix timedelta serialization (#1198)

Co-authored-by: Hugh Wells <hugh.wells@wise.com>

Author: edgarrmondragon

singer-connectors/tap-mysql/tap_mysql/sync_strategies/binlog.py

@@ -224,7 +224,10 @@ def row_to_singer_record(catalog_entry, version, db_column_map, row, time_extrac
         elif isinstance(val, datetime.timedelta):
             if property_format == 'time':
                 # this should convert time column into 'HH:MM:SS' formatted string
-                row_to_persist[column_name] = str(val)
+                _total_seconds = int(val.total_seconds())
+                _hours, _remainder = divmod(_total_seconds, 3600)
+                _minutes, _seconds = divmod(_remainder, 60)
+                row_to_persist[column_name] = f"{_hours:02}:{_minutes:02}:{_seconds:02}"
             else:
                 timedelta_from_epoch = datetime.datetime.utcfromtimestamp(0) + val
                 row_to_persist[column_name] = timedelta_from_epoch.isoformat() + '+00:00'

singer-connectors/tap-mysql/tap_mysql/sync_strategies/common.py

@@ -103,7 +103,10 @@ def row_to_singer_record(catalog_entry, version, row, columns, time_extracted):
 
         elif isinstance(elem, datetime.timedelta):
             if property_format == 'time':
-                row_to_persist += (str(elem),) # this should convert time column into 'HH:MM:SS' formatted string
+                _total_seconds = int(elem.total_seconds())
+                _hours, _remainder = divmod(_total_seconds, 3600)
+                _minutes, _seconds = divmod(_remainder, 60)
+                row_to_persist += (f"{_hours:02}:{_minutes:02}:{_seconds:02}",) # this should convert time column into 'HH:MM:SS' formatted string
             else:
                 epoch = datetime.datetime.utcfromtimestamp(0)
                 timedelta_from_epoch = epoch + elem

singer-connectors/tap-mysql/tests/unit/sync_strategies/test_binlog.py

@@ -2147,3 +2147,29 @@ def test_calculate_gtid_bookmark_for_mysql_no_gtid_found_expect_exception(self):
 
         self.assertEqual("Couldn't find any gtid in state bookmarks to resume logical replication",
                          str(context.exception))
+
+    def test_row_to_singer_record(self):
+        catalog_entry = CatalogEntry(
+            stream='stream',
+            schema=Schema.from_dict({
+                'type': 'object',
+                'properties': {
+                    'time': {
+                        'type': 'string',
+                        'format': 'time',
+                    },
+                },
+            }),
+        )
+        message = binlog.row_to_singer_record(
+            catalog_entry,
+            version=1,
+            row={'time': datetime.timedelta(hours=8, minutes=30)},
+            db_column_map={},
+            time_extracted=datetime.datetime.now(datetime.timezone.utc),
+        )
+
+        assert message.stream == 'stream'
+        assert message.version == 1
+        assert message.record == {'time': '08:30:00'}
+        assert message.time_extracted is not None

singer-connectors/tap-mysql/tests/unit/sync_strategies/test_common.py

@@ -0,0 +1,35 @@
+import datetime
+
+from singer.catalog import CatalogEntry
+from singer.schema import Schema
+
+from tap_mysql.sync_strategies import common
+
+
+class TestCommonSyncStrategyHelpers:
+
+    def test_row_to_singer_record(self):
+        catalog_entry = CatalogEntry(
+            stream='stream',
+            schema=Schema.from_dict({
+                'type': 'object',
+                'properties': {
+                    'time': {
+                        'type': 'string',
+                        'format': 'time',
+                    },
+                },
+            }),
+        )
+        message = common.row_to_singer_record(
+            catalog_entry,
+            version=1,
+            row=(datetime.timedelta(hours=8, minutes=30),),
+            columns=['time'],
+            time_extracted=datetime.datetime.now(datetime.timezone.utc),
+        )
+
+        assert message.stream == 'stream'
+        assert message.version == 1
+        assert message.record == {'time': '08:30:00'}
+        assert message.time_extracted is not None