tessl/pypi-diffsync

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

—

Pending

Overview

Eval results

Files

Model Definition

Name: tessl/pypi-diffsync
Author: tessl

Core functionality for defining data models that represent your domain objects. Models specify unique identifiers, trackable attributes, and parent-child relationships between different object types.

Capabilities

DiffSyncModel Base Class

Base class for all DiffSync object models. Inherits from Pydantic's BaseModel and provides the foundation for defining structured data models with diffing and syncing capabilities.

class DiffSyncModel(BaseModel):
    """Base class for all DiffSync object models."""
    
    _modelname: ClassVar[str] = "diffsyncmodel"
    _identifiers: ClassVar[Tuple[str, ...]] = ()
    _shortname: ClassVar[Tuple[str, ...]] = ()
    _attributes: ClassVar[Tuple[str, ...]] = ()
    _children: ClassVar[Dict[str, str]] = {}
    
    model_flags: DiffSyncModelFlags = DiffSyncModelFlags.NONE
    adapter: Optional["Adapter"] = None
    
    model_config = ConfigDict(arbitrary_types_allowed=True)

Usage Example

from diffsync import DiffSyncModel

class Device(DiffSyncModel):
    _modelname = "device"
    _identifiers = ("name", "site")
    _attributes = ("os_version", "vendor", "model")
    _children = {"interface": "interfaces"}
    
    name: str
    site: str
    os_version: str
    vendor: str
    model: str
    interfaces: List[str] = []

class Interface(DiffSyncModel):
    _modelname = "interface"
    _identifiers = ("device_name", "name")
    _attributes = ("ip_address", "status")
    
    device_name: str
    name: str
    ip_address: str
    status: str

Model Class Variables

Class-level configuration that defines the structure and behavior of DiffSync models.

_modelname: ClassVar[str]

Name of this model, used by DiffSync to store and look up instances. Lowercase by convention, typically corresponds to the class name.

_identifiers: ClassVar[Tuple[str, ...]]

List of model fields which together uniquely identify an instance. This identifier must be globally unique among all instances of this class.

_shortname: ClassVar[Tuple[str, ...]]

Optional list of model fields that together form a shorter identifier. Must be locally unique but doesn't need to be globally unique among all instances.

_attributes: ClassVar[Tuple[str, ...]]

Optional list of additional model fields (beyond identifiers) that are relevant for diff calculation. Only fields in _attributes and _children are considered for diff purposes.

_children: ClassVar[Dict[str, str]]

Optional dictionary mapping child model names to field names for storing child model references. Format: {modelname: field_name}.

Instance Attributes

Runtime attributes that control model behavior and maintain state during operations.

model_flags: DiffSyncModelFlags

Optional behavioral flags for this DiffSyncModel. Can be set as class attribute or instance attribute.

adapter: Optional["Adapter"]

The Adapter instance that owns this model instance. Set automatically when models are added to adapters.

CRUD Operations

Methods for creating, updating, and deleting model instances with platform-specific data operations.

@classmethod
def create(cls, adapter: "Adapter", ids: Dict, attrs: Dict) -> Optional[Self]:
    """
    Instantiate this class with platform-specific data creation.
    
    Args:
        adapter: The master data store for other DiffSyncModel instances
        ids: Dictionary of unique-identifiers needed to create the new object
        attrs: Dictionary of additional attributes to set on the new object
    
    Returns:
        DiffSyncModel instance if creation successful, None if failed
        
    Raises:
        ObjectNotCreated: if an error occurred
    """

def update(self, attrs: Dict) -> Optional[Self]:
    """
    Update the attributes of this instance with platform-specific data updates.
    
    Args:
        attrs: Dictionary of attributes to update on the object
    
    Returns:
        DiffSyncModel instance if update successful, None if failed
        
    Raises:
        ObjectNotUpdated: if an error occurred
    """

def delete(self) -> Optional[Self]:
    """
    Delete any platform-specific data corresponding to this instance.
    
    Returns:
        DiffSyncModel instance if deletion successful, None if failed
        
    Raises:
        ObjectNotDeleted: if an error occurred
    """

CRUD Usage Example

class NetworkDevice(DiffSyncModel):
    _modelname = "device"
    _identifiers = ("name",)
    _attributes = ("ip_address", "os_version")
    
    name: str
    ip_address: str
    os_version: str
    
    @classmethod
    def create(cls, adapter, ids, attrs):
        # Create device in actual network system
        device = super().create(adapter, ids, attrs)
        if device:
            # Perform actual device provisioning
            success = provision_device(device.name, device.ip_address)
            if success:
                device.set_status(DiffSyncStatus.SUCCESS, "Device provisioned")
            else:
                device.set_status(DiffSyncStatus.FAILURE, "Failed to provision")
        return device
    
    def update(self, attrs):
        # Update device in actual network system
        old_ip = self.ip_address
        device = super().update(attrs)
        if device and 'ip_address' in attrs:
            # Perform actual IP address change
            success = update_device_ip(self.name, old_ip, self.ip_address)
            if success:
                device.set_status(DiffSyncStatus.SUCCESS, "IP updated")
            else:
                device.set_status(DiffSyncStatus.FAILURE, "Failed to update IP")
        return device
    
    def delete(self):
        # Delete device from actual network system
        success = deprovision_device(self.name)
        device = super().delete()
        if success:
            device.set_status(DiffSyncStatus.SUCCESS, "Device deprovisioned")
        else:
            device.set_status(DiffSyncStatus.FAILURE, "Failed to deprovision")
        return device

Model Inspection Methods

Methods for retrieving model metadata and current state information.

@classmethod
def get_type(cls) -> str:
    """Return the type/modelname of the object or the class."""

@classmethod
def create_unique_id(cls, **identifiers: Dict[str, Any]) -> str:
    """
    Construct a unique identifier for this model class.
    
    Args:
        **identifiers: Dict of identifiers and their values, as in get_identifiers()
    """

def get_identifiers(self) -> Dict:
    """Get a dict of all identifiers (primary keys) and their values for this object."""

def get_attrs(self) -> Dict:
    """Get all the non-primary-key attributes or parameters for this object."""

def get_unique_id(self) -> str:
    """Get the unique ID of an object."""

def get_shortname(self) -> str:
    """Get the (not guaranteed-unique) shortname of an object, if any."""

def get_status(self) -> Tuple[DiffSyncStatus, str]:
    """Get the status of the last create/update/delete operation on this object, and any associated message."""

Child Relationship Management

Methods for managing parent-child relationships between model instances.

def add_child(self, child: "DiffSyncModel") -> None:
    """
    Add a child reference to an object.
    
    The child object isn't stored, only its unique id.
    The name of the target attribute is defined in _children per object type.
    
    Raises:
        ObjectStoreWrongType: if the type is not part of _children
        ObjectAlreadyExists: if the unique id is already stored
    """

def remove_child(self, child: "DiffSyncModel") -> None:
    """
    Remove a child reference from an object.
    
    Raises:
        ObjectStoreWrongType: if the child model type is not part of _children
        ObjectNotFound: if the child wasn't previously present
    """

Status Management

Methods for tracking and reporting the status of operations performed on model instances.

def set_status(self, status: DiffSyncStatus, message: str = "") -> None:
    """Update the status (and optionally status message) of this model in response to a create/update/delete call."""

Serialization Methods

Methods for converting model instances to various serializable formats.

def dict(self, **kwargs: Any) -> Dict:
    """Convert this DiffSyncModel to a dict, excluding the adapter field by default as it is not serializable."""

def json(self, **kwargs: Any) -> str:
    """Convert this DiffSyncModel to a JSON string, excluding the adapter field by default as it is not serializable."""

def str(self, include_children: bool = True, indent: int = 0) -> str:
    """Build a detailed string representation of this DiffSyncModel and optionally its children."""

Types

from enum import Enum
from typing import ClassVar, Dict, List, Optional, Tuple, Type, Union

class DiffSyncStatus(Enum):
    UNKNOWN = "unknown"
    SUCCESS = "success"
    FAILURE = "failure"
    ERROR = "error"

Install with Tessl CLI