Source Hubspot: cast timestamp to date/datetime

airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

@@ -359,7 +359,38 @@ def read_records(
             yield from []
 
     @staticmethod
-    def _cast_value(declared_field_types: List, field_name: str, field_value: Any, declared_format: str = None) -> Any:
+    def _convert_datetime_to_string(dt: pendulum.datetime, declared_format: str = None) -> str:
+        if declared_format == "date":
+            return dt.to_date_string()
+        elif declared_format == "date-time":
+            return dt.to_datetime_string()
+
+    @classmethod
+    def _cast_datetime(cls, field_name: str, field_value: Any, declared_format: str = None) -> Any:
+        """
+        If format is date/date-time, but actual value is timestamp, convert timestamp to date/date-time string.
+        """
+        if not field_value:
+            return field_value
+
+        try:
+            dt = pendulum.parse(field_value)
+            return cls._convert_datetime_to_string(dt, declared_format=declared_format)
+        except (ValueError, TypeError) as ex:
+            logger.warning(
+                f"Couldn't parse date/datetime string in {field_name}, trying to parse timestamp... Field value: {field_value}. Ex: {ex}"
+            )
+
+        try:
+            dt = pendulum.from_timestamp(int(field_value) / 1000)
+            return cls._convert_datetime_to_string(dt, declared_format=declared_format)
+        except (ValueError, TypeError) as ex:
+            logger.warning(f"Couldn't parse timestamp in {field_name}. Field value: {field_value}. Ex: {ex}")
+
+        return field_value
+
+    @classmethod
+    def _cast_value(cls, declared_field_types: List, field_name: str, field_value: Any, declared_format: str = None) -> Any:
         """
         Convert record's received value according to its declared catalog json schema type / format / attribute name.
         :param declared_field_types type from catalog schema
@@ -377,6 +408,9 @@ def _cast_value(declared_field_types: List, field_name: str, field_value: Any, d
             if declared_format and field_value == "":
                 return None
 
+        if declared_format in ["date", "date-time"]:
+            field_value = cls._cast_datetime(field_name, field_value, declared_format=declared_format)
+
         actual_field_type = type(field_value)
         actual_field_type_name = CUSTOM_FIELD_TYPE_TO_VALUE.get(actual_field_type)
         if actual_field_type_name in declared_field_types:

airbyte-integrations/connectors/source-hubspot/unit_tests/test_field_type_converting.py

@@ -72,7 +72,7 @@ def test_bad_field_type_converting(field_type, expected, caplog, capsys):
         # Test casting fields with format specified
         (["null", "string"], "some_field", "", "date-time", None),
         (["string"], "some_field", "", "date-time", ""),
-        (["null", "string"], "some_field", "2020", "date-time", "2020"),
+        (["null", "string"], "some_field", "2020", "date-time", "2020-01-01 00:00:00"),
     ],
 )
 def test_cast_type_if_needed(declared_field_types, field_name, field_value, format, casted_value):
@@ -82,3 +82,22 @@ def test_cast_type_if_needed(declared_field_types, field_name, field_value, form
         )
         == casted_value
     )
+
+
+@pytest.mark.parametrize(
+    "field_value, declared_format, expected_casted_value",
+    [
+        ("1653696000000", "date", "2022-05-28"),
+        ("1645608465000", "date-time", "2022-02-23 09:27:45"),
+        (1645608465000, "date-time", "2022-02-23 09:27:45"),
+        ("2022-05-28", "date", "2022-05-28"),
+        ("2022-02-23 09:27:45", "date-time", "2022-02-23 09:27:45"),
+        ("", "date", ""),
+        (None, "date", None),
+        ("2022-02-23 09:27:45", "date", "2022-02-23"),
+        ("2022-05-28", "date-time", "2022-05-28 00:00:00"),
+    ],
+)
+def test_cast_timestamp_to_date(field_value, declared_format, expected_casted_value):
+    casted_value = Stream._cast_datetime("hs_recurring_billing_end_date", field_value, declared_format=declared_format)
+    assert casted_value == expected_casted_value