updated duplicates collection

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/parsers/manifest_deduplicator.py

@@ -13,7 +13,7 @@
 # Type definitions for better readability
 ManifestType = Dict[str, Any]
 DefinitionsType = Dict[str, Any]
-DuplicatesType = DefaultDict[str, List[Tuple[List[str], Dict, Dict]]]
+DuplicatesType = DefaultDict[str, List[Tuple[List[str], Dict[str, Any], Dict[str, Any]]]]
 
 # Configuration constants
 N_OCCURANCES = 2
@@ -27,9 +27,6 @@
     "url_base",
 ]
 
-# the placeholder for collected duplicates
-DUPLICATES: DuplicatesType = defaultdict(list, {})
-
 
 def deduplicate_definitions(resolved_manifest: ManifestType) -> ManifestType:
     """
@@ -47,23 +44,23 @@ def deduplicate_definitions(resolved_manifest: ManifestType) -> ManifestType:
         _manifest = copy.deepcopy(resolved_manifest)
         definitions = _manifest.get(DEF_TAG, {})
 
-        _collect_duplicates(definitions)
-        _handle_duplicates(definitions)
+        duplicates = _collect_duplicates(definitions)
+        _handle_duplicates(definitions, duplicates)
 
         return _manifest
     except ManifestDeduplicationException:
         # if any arror occurs, we just return the original manifest.
         return resolved_manifest
 
 
-def _replace_duplicates_with_refs(definitions: ManifestType) -> None:
+def _replace_duplicates_with_refs(definitions: ManifestType, duplicates: DuplicatesType) -> None:
     """
     Process duplicate objects and replace them with references.
 
     Args:
         definitions: The definitions dictionary to modify
     """
-    for _, occurrences in DUPLICATES.items():
+    for _, occurrences in duplicates.items():
         # Skip non-duplicates
         if len(occurrences) < N_OCCURANCES:
             continue
@@ -84,20 +81,20 @@ def _replace_duplicates_with_refs(definitions: ManifestType) -> None:
                 parent_obj[key] = _create_ref_object(ref_key)
 
 
-def _handle_duplicates(definitions: DefinitionsType) -> None:
+def _handle_duplicates(definitions: DefinitionsType, duplicates: DuplicatesType) -> None:
     """
-    Process the DUPLICATES and replace them with references.
+    Process the duplicates and replace them with references.
 
     Args:
-        DUPLICATES: Dictionary of duplicate objects
+        duplicates: Dictionary of duplicate objects
     """
     # process duplicates only if there are any
-    if len(DUPLICATES) > 0:
+    if len(duplicates) > 0:
         if not SHARED_TAG in definitions:
             definitions[SHARED_TAG] = {}
 
         try:
-            _replace_duplicates_with_refs(definitions)
+            _replace_duplicates_with_refs(definitions, duplicates)
         except Exception as e:
             raise ManifestDeduplicationException(str(e))
 
@@ -116,19 +113,21 @@ def _is_allowed_tag(key: str) -> bool:
 
 
 def _add_duplicate(
+    duplicates: DuplicatesType,
     current_path: List[str],
-    obj: Dict,
+    obj: Dict[str, Any],
     value: Any,
     key: Optional[str] = None,
 ) -> None:
     """
     Adds a duplicate record of an observed object by computing a unique hash for the provided value.
 
     This function computes a hash for the given value (or a dictionary composed of the key and value if a key is provided)
-    and appends a tuple containing the current path, the original object, and the value to the global DUPLICATES
+    and appends a tuple containing the current path, the original object, and the value to the duplicates
     dictionary under the corresponding hash.
 
     Parameters:
+        duplicates (DuplicatesType): The dictionary to store duplicate records.
         current_path (List[str]): The list of keys or indices representing the current location in the object hierarchy.
         obj (Dict): The original dictionary object where the duplicate is observed.
         value (Any): The value to be hashed and used for identifying duplicates.
@@ -138,7 +137,7 @@ def _add_duplicate(
     value_to_hash = value if key is None else {key: value}
     obj_hash = _hash_object(value_to_hash)
     if obj_hash:
-        DUPLICATES[obj_hash].append((current_path, obj, value))
+        duplicates[obj_hash].append((current_path, obj, value))
 
 
 def _add_to_shared_definitions(
@@ -161,49 +160,61 @@ def _add_to_shared_definitions(
     return definitions
 
 
-def _collect_duplicates(node: ManifestType, path: Optional[List[str]] = None) -> None:
+def _collect_duplicates(node: ManifestType) -> DuplicatesType:
     """
     Traverse the JSON object and collect all potential duplicate values and objects.
 
     Args:
-        node: The JSON object to analyze
+        node: The JSON object to analyze.
 
     Returns:
-        DUPLICATES: A dictionary of duplicate objects
+        duplicates: A dictionary of duplicate objects.
     """
 
-    try:
-        if not isinstance(node, dict):
+    def _collect(obj: Dict[str, Any], path: Optional[List[str]] = None) -> None:
+        """
+        The closure to recursively collect duplicates in the JSON object.
+
+        Args:
+            obj: The current object being analyzed.
+            path: The current path in the object hierarchy.
+        """
+        if not isinstance(obj, dict):
             return
 
         path = [] if path is None else path
-
         # Check if the object is empty
-        for key, value in node.items():
+        for key, value in obj.items():
             current_path = path + [key]
 
             if isinstance(value, dict):
                 # First process nested dictionaries
-                _collect_duplicates(value, current_path)
+                _collect(value, current_path)
                 # Process allowed-only component tags
                 if _is_allowed_tag(key):
-                    _add_duplicate(current_path, node, value)
+                    _add_duplicate(duplicates, current_path, obj, value)
 
             # handle primitive types
             elif isinstance(value, (str, int, float, bool)):
                 # Process allowed-only field tags
                 if _is_allowed_tag(key):
-                    _add_duplicate(current_path, node, value, key)
+                    _add_duplicate(duplicates, current_path, obj, value, key)
 
             # handle list cases
             elif isinstance(value, list):
                 for i, item in enumerate(value):
-                    _collect_duplicates(item, current_path + [str(i)])
+                    _collect(item, current_path + [str(i)])
+
+    duplicates: DuplicatesType = defaultdict(list, {})
+
+    try:
+        _collect(node)
+        return duplicates
     except Exception as e:
         raise ManifestDeduplicationException(str(e))
 
 
-def _hash_object(node: Dict) -> Optional[str]:
+def _hash_object(node: Dict[str, Any]) -> Optional[str]:
     """
     Create a unique hash for a dictionary object.
 

unit_tests/connector_builder/test_connector_builder_handler.py

@@ -384,6 +384,7 @@ def test_resolve_manifest(valid_resolve_manifest_config_file):
                 },
                 "record_selector": {"extractor": {"field_path": ["result"]}},
             },
+            "shared": {},
         },
         "streams": [
             {

unit_tests/sources/declarative/interpolation/test_macros.py

@@ -5,6 +5,7 @@
 import datetime
 
 import pytest
+from freezegun import freeze_time
 
 from airbyte_cdk.sources.declarative.interpolation.macros import macros
 
@@ -29,6 +30,7 @@ def test_macros_export(test_name, fn_name, found_in_macros):
         assert fn_name not in macros
 
 
+@freeze_time("2022-01-01")
 @pytest.mark.parametrize(
     "input_value, format, input_format, expected_output",
     [