feat(source-hubspot): Migrate deals_archived, forms, form_submissions, owners, owners_archived to low-code

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

@@ -2,12 +2,17 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 #
 
-from typing import Any, Dict, Optional
+from dataclasses import InitVar, dataclass
+from typing import Any, Dict, List, Mapping, Optional, Union
 
+import dpath
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.transformations import RecordTransformation
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 
+@dataclass
 class NewtoLegacyFieldTransformation(RecordTransformation):
     """
     Implements a custom transformation which adds the legacy field equivalent of v2 fields for streams which contain Deals and Contacts entities.
@@ -18,8 +23,7 @@ class NewtoLegacyFieldTransformation(RecordTransformation):
     hs_v2_date_exited_{stage_id} -> hs_date_exited_{stage_id} where {stage_id} is a user-generated value
     """
 
-    def __init__(self, field_mapping: Dict[str, str]) -> None:
-        self._field_mapping = field_mapping
+    field_mapping: Mapping[str, str]
 
     def transform(
         self,
@@ -36,7 +40,7 @@ def transform(
         is_record = record_or_schema.get("properties") is not None
 
         for field, value in list(record_or_schema.get("properties", record_or_schema).items()):
-            for legacy_field, new_field in self._field_mapping.items():
+            for legacy_field, new_field in self.field_mapping.items():
                 if new_field in field:
                     transformed_field = field.replace(new_field, legacy_field)