Merge branch 'main' into copilot/fix-80

Author: BeArchiTek

.github/workflows/sync-docs.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout source repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           path: source-repo
 

.github/workflows/trigger-push-stable.yml

@@ -45,7 +45,7 @@ jobs:
       - name: "Install Poetry"
         uses: "snok/install-poetry@v1"
         with:
-          version: 1.8.5
+          version: 2.1.3
           virtualenvs-create: true
           virtualenvs-in-project: true
           installer-parallel: true
@@ -74,7 +74,7 @@ jobs:
           poetry version $VERSION  # directly use the VERSION
 
       - name: Update lock file
-        run: poetry lock --no-update
+        run: poetry lock
 
       - name: commit pyproject.toml
         uses: github-actions-x/commit@v2.9

.github/workflows/workflow-linter.yml

@@ -63,7 +63,7 @@ jobs:
           - "3.11"
           - "3.12"
         poetry-version:
-          - "1.8.5"
+          - "2.1.3"
     timeout-minutes: 5
     steps:
       - name: "Check out repository code"

.github/workflows/workflow-poetry-check.yml

@@ -37,7 +37,7 @@ jobs:
         python-version:
           - "3.12"
         poetry-version:
-          - "1.8.5"
+          - "2.1.3"
     steps:
       - name: "Check out repository code"
         uses: "actions/checkout@v4"

.github/workflows/workflow-publish.yml

@@ -47,7 +47,7 @@ jobs:
         python-version:
           - "3.12"
         poetry-version:
-          - "1.8.5"
+          - "2.1.3"
     steps:
       - name: "Set up Python ${{ matrix.python-version }}"
         uses: "actions/setup-python@v5"

docs/package-lock.json

@@ -21,6 +21,7 @@
     SyncAdapter,
     SyncConfig,
 )
+from infrahub_sync.adapters.utils import build_mapping
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -45,41 +46,14 @@ def __init__(self, target: str, adapter: SyncAdapter, config: SyncConfig, *args,
     def _create_ipfabric_client(self, adapter: SyncAdapter) -> IPFClient:
         settings = adapter.settings or {}
 
-        base_url = settings.get("base_url") or None
-        auth = settings.get("auth") or None
-        timeout = settings.get("timeout", 10)
-        verify_ssl = settings.get("verify_ssl", True)
+        base_url = settings.get("base_url", None)
+        auth = settings.get("auth", None)
 
         if not base_url or not auth:
             msg = "Both url and auth must be specified!"
             raise ValueError(msg)
 
-        return IPFClient(base_url=base_url, auth=auth, timeout=timeout, verify=verify_ssl)
-
-    def build_mapping(self, reference, obj) -> str:
-        # Get object class and model name from the store
-        object_class, modelname = self.store._get_object_class_and_model(model=reference)
-
-        # Find the schema element matching the model name
-        schema_element = next(
-            (element for element in self.config.schema_mapping if element.name == modelname),
-            None,
-        )
-
-        # Collect all relevant field mappings for identifiers
-        new_identifiers = []
-
-        # Convert schema_element.fields to a dictionary for fast lookup
-        field_dict = {field.name: field.mapping for field in schema_element.fields}
-
-        # Loop through object_class._identifiers to find corresponding field mappings
-        for identifier in object_class._identifiers:
-            if identifier in field_dict:
-                new_identifiers.append(field_dict[identifier])
-
-        # Construct the unique identifier, using a fallback if a key isn't found
-        unique_id = "__".join(str(obj.get(key, "")) for key in new_identifiers)
-        return unique_id
+        return IPFClient(**settings)
 
     def model_loader(self, model_name: str, model: IpfabricsyncModel) -> None:
         """
@@ -143,7 +117,7 @@ def ipfabric_dict_to_diffsync(self, obj: dict, mapping: SchemaMappingModel, mode
                     raise IndexError(msg)
                 if not field_is_list and (node := obj[field.mapping]):
                     matching_nodes = []
-                    node_id = self.build_mapping(reference=field.reference, obj=obj)
+                    node_id = build_mapping(adapter=self, reference=field.reference, obj=obj, field=field)
                     matching_nodes = [item for item in nodes if str(item) == node_id]
                     if len(matching_nodes) == 0:
                         data[field.name] = None

infrahub_sync/adapters/ipfabricsync.py

@@ -18,8 +18,7 @@
     SyncAdapter,
     SyncConfig,
 )
-
-from .utils import get_value
+from infrahub_sync.adapters.utils import build_mapping, get_value
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -220,32 +219,6 @@ def model_loader(self, model_name: str, model: SlurpitsyncModel) -> None:
         if self.skipped:
             print(f"{self.type}: skipped syncing {len(self.skipped)} models")
 
-    # Reuse mapping from another adapter
-    def build_mapping(self, reference, obj):
-        # Get object class and model name from the store
-        object_class, modelname = self.store._get_object_class_and_model(model=reference)
-
-        # Find the schema element matching the model name
-        schema_element = next(
-            (element for element in self.config.schema_mapping if element.name == modelname),
-            None,
-        )
-
-        # Collect all relevant field mappings for identifiers
-        new_identifiers = []
-
-        # Convert schema_element.fields to a dictionary for fast lookup
-        field_dict = {field.name: field.mapping for field in schema_element.fields}
-
-        # Loop through object_class._identifiers to find corresponding field mappings
-        for identifier in object_class._identifiers:
-            if identifier in field_dict:
-                new_identifiers.append(field_dict[identifier])
-
-        # Construct the unique identifier, using a fallback if a key isn't found
-        unique_id = "__".join(str(obj.get(key, "")) for key in new_identifiers)
-        return unique_id
-
     def slurpit_obj_to_diffsync(
         self, obj: dict[str, Any], mapping: SchemaMappingModel, model: SlurpitsyncModel
     ) -> dict:
@@ -276,7 +249,7 @@ def slurpit_obj_to_diffsync(
                 if not field_is_list:
                     if node := obj.get(field.mapping):
                         matching_nodes = []
-                        node_id = self.build_mapping(reference=field.reference, obj=obj)
+                        node_id = build_mapping(adapter=self, reference=field.reference, obj=obj)
                         matching_nodes = [item for item in nodes if str(item) == node_id]
                         if len(matching_nodes) == 0:
                             self.skipped.append(node)

infrahub_sync/adapters/slurpitsync.py

@@ -1,6 +1,42 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from diffsync import Adapter
+
+
+def build_mapping(adapter: Adapter, reference: str, obj, field) -> str:
+    """This is used when references are encountered to attempt to resolve them for mapping."""
+    # Get object class and model name from the store
+    object_class, modelname = adapter.store._get_object_class_and_model(model=reference)
+
+    # Find the schema element matching the model name
+    schema_element = next(
+        (element for element in adapter.config.schema_mapping if element.name == modelname),
+        None,
+    )
+    if not schema_element:
+        msg = (
+            f"Schema mapping for model '{reference}' not found when attempting to resolve "
+            f"reference for {field.name}. The reference must be an existing schema mapping."
+        )
+        raise ValueError(msg)
+
+    # Collect all relevant field mappings for identifiers
+    new_identifiers = []
+
+    # Convert schema_element.fields to a dictionary for fast lookup
+    field_dict = {field.name: field.mapping for field in schema_element.fields}
+
+    # Loop through object_class._identifiers to find corresponding field mappings
+    for identifier in object_class._identifiers:
+        if identifier in field_dict:
+            new_identifiers.append(field_dict[identifier])
+
+    # Construct the unique identifier, using a fallback if a key isn't found
+    unique_id = "__".join(str(obj.get(key, "")) for key in new_identifiers)
+    return unique_id
 
 
 def get_value(obj: Any, name: str) -> Any | None: