Upgrading to Pydantic v2 and DiffSync v2

network_importer/adapters/base.py

@@ -1,10 +1,10 @@
 """BaseAdapter for the network importer."""
-from diffsync import DiffSync
+from diffsync import Adapter
 from diffsync.exceptions import ObjectNotFound
 from network_importer.models import Site, Device, Interface, IPAddress, Cable, Vlan, Prefix
 
 
-class BaseAdapter(DiffSync):
+class BaseAdapter(Adapter):
     """Base Adapter for the network importer."""
 
     site = Site

network_importer/adapters/nautobot_api/adapter.py

@@ -201,7 +201,7 @@ def load_nautobot_vlan(self, site):
         vlans = self.nautobot.ipam.vlans.filter(site=site.name)
 
         for nb_vlan in vlans:
-            vlan = self.vlan.create_from_pynautobot(diffsync=self, obj=nb_vlan, site_name=site.name)
+            vlan = self.vlan.create_from_pynautobot(adapter=self, obj=nb_vlan, site_name=site.name)
             self.add(vlan)
             site.add_child(vlan)
 
@@ -326,7 +326,7 @@ def load_nautobot_ip_address(self, site, device):  # pylint: disable=unused-argu
 
         ips = self.nautobot.ipam.ip_addresses.filter(device=device.name)
         for ipaddr in ips:
-            ip_address = self.ip_address.create_from_pynautobot(diffsync=self, obj=ipaddr, device_name=device.name)
+            ip_address = self.ip_address.create_from_pynautobot(adapter=self, obj=ipaddr, device_name=device.name)
             ip_address, _ = self.get_or_add(ip_address)
 
             interface = self.get(

network_importer/adapters/nautobot_api/settings.py

@@ -1,6 +1,7 @@
 """Settings definition for the NetboxAPIAdapter."""
 from typing import List, Optional, Union
-from pydantic import BaseSettings
+from pydantic_settings import BaseSettings
+from pydantic import Field
 
 from diffsync import DiffSyncModelFlags
 
@@ -9,25 +10,16 @@ class AdapterSettings(BaseSettings):
     """Config settings for the netbox_api adapter. Not used currently."""
 
     model_flag_tags: List[str] = list()  # List of tags that defines what objects to assign the model_flag to.
-    model_flag: Optional[DiffSyncModelFlags]  # The model flag that will be applied to objects based on tag.
+    model_flag: Optional[DiffSyncModelFlags] = None  # The model flag that will be applied to objects based on tag.
 
 
 class InventorySettings(BaseSettings):
     """Config settings for the NautobotAPI inventory."""
 
-    address: Optional[str] = "http://localhost"
-    token: Optional[str] = None
-    verify_ssl: Union[bool, str] = True
+    address: Optional[str] = Field(default="http://localhost", env="NAUTOBOT_ADDRESS")
+    token: Optional[str] = Field(default=None, env="NAUTOBOT_TOKEN")
+    verify_ssl: Union[bool, str] = Field(default=True, env="NAUTOBOT_VERIFY_SSL")
 
     use_primary_ip: Optional[bool] = True
     fqdn: Optional[str] = None
     filter: Optional[str] = None
-
-    class Config:
-        """Additional parameters to automatically map environment variable to some settings."""
-
-        fields = {
-            "address": {"env": "NAUTOBOT_ADDRESS"},
-            "token": {"env": "NAUTOBOT_TOKEN"},
-            "verify_ssl": {"env": "NAUTOBOT_VERIFY_SSL"},
-        }

network_importer/adapters/netbox_api/adapter.py

@@ -208,7 +208,7 @@ def load_netbox_vlan(self, site):
         vlans = self.netbox.ipam.vlans.filter(site=site.name)
 
         for nb_vlan in vlans:
-            vlan = self.vlan.create_from_pynetbox(diffsync=self, obj=nb_vlan, site_name=site.name)
+            vlan = self.vlan.create_from_pynetbox(adapter=self, obj=nb_vlan, site_name=site.name)
             self.add(vlan)
             site.add_child(vlan)
 
@@ -333,7 +333,7 @@ def load_netbox_ip_address(self, site, device):  # pylint: disable=unused-argume
 
         ips = self.netbox.ipam.ip_addresses.filter(device=device.name)
         for ipaddr in ips:
-            ip_address = self.ip_address.create_from_pynetbox(diffsync=self, obj=ipaddr, device_name=device.name)
+            ip_address = self.ip_address.create_from_pynetbox(adapter=self, obj=ipaddr, device_name=device.name)
             ip_address, _ = self.get_or_add(ip_address)
 
             interface = self.get(