Source Hubspot: cast timestamp to date/datetime

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -328,7 +328,7 @@
 - name: HubSpot
   sourceDefinitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
   dockerRepository: airbyte/source-hubspot
-  dockerImageTag: 0.1.42
+  dockerImageTag: 0.1.43
   documentationUrl: https://docs.airbyte.io/integrations/sources/hubspot
   icon: hubspot.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -3148,7 +3148,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-hubspot:0.1.42"
+- dockerImage: "airbyte/source-hubspot:0.1.43"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/hubspot"
     connectionSpecification:

airbyte-integrations/connectors/source-hubspot/Dockerfile

@@ -34,5 +34,5 @@ COPY source_hubspot ./source_hubspot
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.42
+LABEL io.airbyte.version=0.1.43
 LABEL io.airbyte.name=airbyte/source-hubspot

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

docs/integrations/sources/hubspot.md

@@ -112,6 +112,7 @@ If you are using Oauth, most of the streams require the appropriate [scopes](htt
 
 | Version | Date       | Pull Request | Subject                                                                                                                                        |
 |:--------|:-----------| :--- |:-----------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.1.43  | 2022-02-24 | [10576](https://github.com/airbytehq/airbyte/pull/10576) | Cast timestamp to date/datetime|
 | 0.1.42  | 2022-02-22 | [10492](https://github.com/airbytehq/airbyte/pull/10492) | Add `date-time` format to datetime fields|
 | 0.1.41  | 2022-02-21 | [10177](https://github.com/airbytehq/airbyte/pull/10177) | Migrate to CDK |
 | 0.1.40  | 2022-02-10 | [10142](https://github.com/airbytehq/airbyte/pull/10142) | Add associations to ticket stream |