bug: Allow optional mapping

infrahub_sync/__init__.py

@@ -45,7 +45,7 @@ class SchemaMappingField(pydantic.BaseModel):
 
 class SchemaMappingModel(pydantic.BaseModel):
     name: str
-    mapping: str
+    mapping: str | None = pydantic.Field(default=None)
     identifiers: list[str] | None = pydantic.Field(default=None)
     filters: list[SchemaMappingFilter] | None = pydantic.Field(default=None)
     transforms: list[SchemaMappingTransform] | None = pydantic.Field(default=None)

infrahub_sync/adapters/aci.py

@@ -248,6 +248,10 @@ def model_loader(self, model_name: str, model: builtins.type[AciModel]) -> None:
                 logger.debug("No fields defined for schema mapping %s, skipping", element.name)
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             # Use the resource endpoint from the schema mapping
             resource_name = element.mapping
 

infrahub_sync/adapters/genericrestapi.py

@@ -141,6 +141,10 @@ def model_loader(self, model_name: str, model: GenericrestapiModel) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             # Use the resource endpoint from the schema mapping
             resource_name = element.mapping
 

infrahub_sync/adapters/infrahub.py

@@ -69,6 +69,8 @@ def resolve_peer_node(
         peer_node = client.get(id=key, kind=rel_schema.peer, populate_store=True)
         if not peer_node:
             print(f"Unable to find {rel_schema.peer} [{key}] - Ignored")
+        else:
+            print(f"Found it with kind {peer_node.get_kind()} using Infrahub client")
     return peer_node
 
 

infrahub_sync/adapters/ipfabricsync.py

@@ -58,6 +58,7 @@ def _create_ipfabric_client(self, adapter: SyncAdapter) -> IPFClient:
             auth = os.environ.get("IPF_TOKEN", None)
             settings["auth"] = auth
 
+
         if not base_url or not auth:
             msg = "Both url and auth must be specified! Please specify in the config or using `IPF_URL` and `IPF_TOKEN` environment variables."
             raise ValueError(msg)
@@ -74,11 +75,12 @@ def model_loader(self, model_name: str, model: IpfabricsyncModel) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
             table = self.client.fetch_all(element.mapping, filters=ipf_filters.get(element.mapping))
-            print(f"{self.type}: Loading {len(table)} from `{element.mapping}`")
 
             total = len(table)
-
             if self.config.source.name.title() == self.type.title():
                 # Filter records
                 filtered_objs = model.filter_records(records=table, schema_mapping=element)
@@ -90,9 +92,7 @@ def model_loader(self, model_name: str, model: IpfabricsyncModel) -> None:
                 transformed_objs = table
 
             for obj in transformed_objs:
-                print(f"Object to load: {obj}")
                 data = self.ipfabric_dict_to_diffsync(obj=obj, mapping=element, model=model)
-                print(f"Data to load: {data}")
                 item = model(**data)
                 self.update_or_add_model_instance(item)
 

infrahub_sync/adapters/nautobot.py

@@ -61,6 +61,10 @@ def model_loader(self, model_name: str, model: NautobotModel) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             # Use the resource endpoint from the schema mapping
             app_name, resource_name = element.mapping.split(".")
             nautobot_app = getattr(self.client, app_name)

infrahub_sync/adapters/netbox.py

@@ -65,6 +65,10 @@ def model_loader(self, model_name: str, model: NetboxModel) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             # Use the resource endpoint from the schema mapping
             app_name, resource_name = element.mapping.split(".")
             netbox_app = getattr(self.client, app_name)

infrahub_sync/adapters/prometheus.py

@@ -470,6 +470,10 @@ def model_loader(self, model_name: str, model) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             metric_or_resource = element.mapping
             objs = samples_by_metric.get(metric_or_resource, [])
             total = len(objs)

infrahub_sync/adapters/slurpitsync.py

@@ -175,6 +175,10 @@ def model_loader(self, model_name: str, model: SlurpitsyncModel) -> None:
             if element.name != model_name:
                 continue
 
+            if not element.mapping:
+                print(f"No mapping defined for '{element.name}', skipping...")
+                continue
+
             if element.mapping.startswith("planning_results"):
                 planning_name = element.mapping.split(".")[1]
                 nodes = self.planning_results(planning_name)