tessl/pypi-diffsync

Library to easily sync/diff/update 2 different data sources

—

Pending

Overview

Eval results

Files

Data Management

Name: tessl/pypi-diffsync
Author: tessl

Adapter functionality for managing collections of models, loading data from various sources, and providing query and storage operations through configurable storage backends.

Capabilities

Adapter Class

Container class for storing groups of DiffSyncModel instances and performing diff/sync operations between different data sources.

class Adapter:
    """Class for storing a group of DiffSyncModel instances and diffing/synchronizing to another Adapter instance."""
    
    type: Optional[str] = None
    top_level: ClassVar[List[str]] = []
    
    def __init__(self, name: Optional[str] = None,
                 internal_storage_engine: Union[Type[BaseStore], BaseStore] = LocalStore) -> None:
        """
        Generic initialization function.
        
        Args:
            name: Optional name for this adapter instance
            internal_storage_engine: Storage backend class or instance to use
        """

Usage Example

from diffsync import Adapter, DiffSyncModel, LocalStore

class Device(DiffSyncModel):
    _modelname = "device"
    _identifiers = ("name",)
    _attributes = ("vendor", "model")
    
    name: str
    vendor: str
    model: str

class NetworkAdapter(Adapter):
    device = Device  # Map model name to class
    top_level = ["device"]  # Top-level models for traversal
    
    def load(self):
        # Load data from your source
        devices_data = fetch_devices_from_api()
        for device_data in devices_data:
            device = Device(**device_data)
            self.add(device)

# Create adapter instances
source_net = NetworkAdapter(name="source_network")
source_net.load()

Data Loading

Methods for populating adapters with data from various sources.

def load(self) -> None:
    """Load all desired data from whatever backend data source into this instance."""

def load_from_dict(self, data: Dict) -> None:
    """
    The reverse of dict method, taking a dictionary and loading into the inventory.
    
    Args:
        data: Dictionary in the format that dict would export as
    """

Loading Example

class DatabaseAdapter(Adapter):
    device = Device
    interface = Interface
    top_level = ["device"]
    
    def load(self):
        # Load from database
        with get_db_connection() as conn:
            # Load devices
            devices = conn.execute("SELECT * FROM devices").fetchall()
            for device_row in devices:
                device = Device(
                    name=device_row['name'],
                    vendor=device_row['vendor'],
                    model=device_row['model']
                )
                self.add(device)
            
            # Load interfaces
            interfaces = conn.execute("SELECT * FROM interfaces").fetchall()
            for intf_row in interfaces:
                interface = Interface(
                    device_name=intf_row['device_name'],
                    name=intf_row['name'],
                    ip_address=intf_row['ip_address']
                )
                # Add interface to its parent device
                device = self.get("device", intf_row['device_name'])
                device.add_child(interface)
                self.add(interface)

Object Storage Operations

Methods for adding, retrieving, updating, and removing model instances from the adapter.

def add(self, obj: DiffSyncModel) -> None:
    """
    Add a DiffSyncModel object to the store.
    
    Args:
        obj: Object to store
    
    Raises:
        ObjectAlreadyExists: if a different object with the same uid is already present
    """

def update(self, obj: DiffSyncModel) -> None:
    """
    Update a DiffSyncModel object to the store.
    
    Args:
        obj: Object to store
    
    Raises:
        ObjectAlreadyExists: if a different object with the same uid is already present
    """

def remove(self, obj: DiffSyncModel, remove_children: bool = False) -> None:
    """
    Remove a DiffSyncModel object from the store.
    
    Args:
        obj: object to remove
        remove_children: If True, also recursively remove any children of this object
    
    Raises:
        ObjectNotFound: if the object is not present
    """

Object Retrieval

Methods for querying and retrieving stored model instances.

def get(self, obj: Union[str, DiffSyncModel, Type[DiffSyncModel]],
        identifier: Union[str, Dict]) -> DiffSyncModel:
    """
    Get one object from the data store based on its unique id.
    
    Args:
        obj: DiffSyncModel class or instance, or modelname string, that defines the type of the object to retrieve
        identifier: Unique ID of the object to retrieve, or dict of unique identifier keys/values
    
    Raises:
        ValueError: if obj is a str and identifier is a dict (can't convert dict into a uid str without a model class)
        ObjectNotFound: if the requested object is not present
    """

def get_or_none(self, obj: Union[str, DiffSyncModel, Type[DiffSyncModel]],
                identifier: Union[str, Dict]) -> Optional[DiffSyncModel]:
    """
    Get one object from the data store based on its unique id or get a None.
    
    Args:
        obj: DiffSyncModel class or instance, or modelname string, that defines the type of the object to retrieve
        identifier: Unique ID of the object to retrieve, or dict of unique identifier keys/values
    
    Returns:
        DiffSyncModel matching provided criteria or None if not found
    """

def get_all(self, obj: Union[str, DiffSyncModel, Type[DiffSyncModel]]) -> List[DiffSyncModel]:
    """
    Get all objects of a given type.
    
    Args:
        obj: DiffSyncModel class or instance, or modelname string, that defines the type of the objects to retrieve
    
    Returns:
        List of Object
    """

def get_by_uids(self, uids: List[str],
                obj: Union[str, DiffSyncModel, Type[DiffSyncModel]]) -> List[DiffSyncModel]:
    """
    Get multiple objects from the store by their unique IDs/Keys and type.
    
    Args:
        uids: List of unique id / key identifying object in the database
        obj: DiffSyncModel class or instance, or modelname string, that defines the type of the objects to retrieve
    
    Raises:
        ObjectNotFound: if any of the requested UIDs are not found in the store
    """

Retrieval Examples

# Get specific device by name
device = adapter.get("device", "router1")
# Or using dict identifier
device = adapter.get(Device, {"name": "router1"})

# Get all devices
all_devices = adapter.get_all("device")

# Get multiple devices by IDs
device_ids = ["router1", "switch1", "firewall1"]
devices = adapter.get_by_uids(device_ids, "device")

# Safe get that returns None if not found
device = adapter.get_or_none("device", "nonexistent")

Advanced Retrieval Operations

Convenience methods for more complex retrieval and management patterns.

def get_or_instantiate(self, model: Type[DiffSyncModel], ids: Dict,
                      attrs: Optional[Dict] = None) -> Tuple[DiffSyncModel, bool]:
    """
    Attempt to get the object with provided identifiers or instantiate and add it with provided identifiers and attrs.
    
    Args:
        model: The DiffSyncModel to get or create
        ids: Identifiers for the DiffSyncModel to get or create with
        attrs: Attributes when creating an object if it doesn't exist
    
    Returns:
        Tuple of (existing or new object, whether it was created)
    """

def get_or_add_model_instance(self, obj: DiffSyncModel) -> Tuple[DiffSyncModel, bool]:
    """
    Attempt to get the object with provided obj identifiers or add obj.
    
    Args:
        obj: An obj of the DiffSyncModel to get or add
    
    Returns:
        Tuple of (existing or new object, whether it was created)
    """

def update_or_instantiate(self, model: Type[DiffSyncModel], ids: Dict,
                         attrs: Dict) -> Tuple[DiffSyncModel, bool]:
    """
    Attempt to update an existing object with provided ids/attrs or instantiate it with provided identifiers and attrs.
    
    Args:
        model: The DiffSyncModel to update or create
        ids: Identifiers for the DiffSyncModel to update or create with
        attrs: Attributes when creating/updating an object if it doesn't exist
    
    Returns:
        Tuple of (existing or new object, whether it was created)
    """

def update_or_add_model_instance(self, obj: DiffSyncModel) -> Tuple[DiffSyncModel, bool]:
    """
    Attempt to update an existing object with provided obj ids/attrs or instantiate obj.
    
    Args:
        obj: An instance of the DiffSyncModel to update or create
    
    Returns:
        Tuple of (existing or new object, whether it was created)
    """

Information and Statistics

Methods for getting information about the adapter's contents and structure.

def get_all_model_names(self) -> Set[str]:
    """
    Get all model names.
    
    Returns:
        List of model names
    """

def count(self, model: Union[str, "DiffSyncModel", Type["DiffSyncModel"], None] = None) -> int:
    """
    Count how many objects of one model type exist in the backend store.
    
    Args:
        model: The DiffSyncModel to check the number of elements. If not provided, default to all
    
    Returns:
        Number of elements of the model type
    """

@classmethod
def get_tree_traversal(cls, as_dict: bool = False) -> Union[str, Dict]:
    """
    Get a string describing the tree traversal for the diffsync object.
    
    Args:
        as_dict: Whether to return as a dictionary
    
    Returns:
        A string or dictionary representation of tree
    """

Serialization

Methods for converting adapter contents to various serializable formats.

def dict(self, exclude_defaults: bool = True, **kwargs: Any) -> Dict[str, Dict[str, Dict]]:
    """Represent the DiffSync contents as a dict, as if it were a Pydantic model."""

def str(self, indent: int = 0) -> str:
    """Build a detailed string representation of this Adapter."""

Serialization Example

# Convert adapter to dictionary
adapter_data = adapter.dict()

# Convert to string representation
adapter_string = adapter.str()

# Save and load adapter state
import json
with open('adapter_state.json', 'w') as f:
    json.dump(adapter_data, f)

# Load back into new adapter
with open('adapter_state.json', 'r') as f:
    data = json.load(f)

new_adapter = NetworkAdapter()
new_adapter.load_from_dict(data)

Types

from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union, ClassVar
from diffsync.store import BaseStore, LocalStore

# Storage engine type for adapter initialization
StorageEngine = Union[Type[BaseStore], BaseStore]

Install with Tessl CLI