tessl/pypi-diffsync

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

—

Pending

Overview

Eval results

Files

Flags and Configuration

Name: tessl/pypi-diffsync
Author: tessl

Behavioral control flags and configuration options for customizing diff calculation and synchronization behavior, including error handling, skipping patterns, and logging verbosity.

Capabilities

DiffSyncFlags

Flags that can be passed to sync_* or diff_* calls to affect their behavior.

class DiffSyncFlags(enum.Flag):
    """Flags that can be passed to a sync_* or diff_* call to affect its behavior."""
    
    NONE = 0
    CONTINUE_ON_FAILURE = 0b1
    SKIP_UNMATCHED_SRC = 0b10
    SKIP_UNMATCHED_DST = 0b100
    SKIP_UNMATCHED_BOTH = SKIP_UNMATCHED_SRC | SKIP_UNMATCHED_DST
    LOG_UNCHANGED_RECORDS = 0b1000

Flag Descriptions

NONE = 0

No special flags - default behavior.

CONTINUE_ON_FAILURE = 0b1

Continue synchronizing even if failures are encountered when syncing individual models. Without this flag, any failure will stop the entire sync operation.

SKIP_UNMATCHED_SRC = 0b10

Ignore objects that only exist in the source/"from" DiffSync when determining diffs and syncing. If this flag is set, no new objects will be created in the target/"to" DiffSync.

SKIP_UNMATCHED_DST = 0b100

Ignore objects that only exist in the target/"to" DiffSync when determining diffs and syncing. If this flag is set, no objects will be deleted from the target/"to" DiffSync.

SKIP_UNMATCHED_BOTH = SKIP_UNMATCHED_SRC | SKIP_UNMATCHED_DST

Combination flag - ignore objects that exist only in source OR only in target. Only update existing objects that are present in both.

LOG_UNCHANGED_RECORDS = 0b1000

If this flag is set, a log message will be generated during synchronization for each model, even unchanged ones. By default, only models that have actual changes to synchronize will be logged. This flag is off by default to reduce verbosity but can be enabled for debugging.

DiffSyncFlags Usage Examples

from diffsync import DiffSyncFlags

# Basic synchronization with default behavior
diff = target.sync_from(source)

# Continue sync even if some operations fail
diff = target.sync_from(source, flags=DiffSyncFlags.CONTINUE_ON_FAILURE)

# Only update existing objects, don't create new ones
diff = target.sync_from(source, flags=DiffSyncFlags.SKIP_UNMATCHED_SRC)

# Only update existing objects, don't delete missing ones
diff = target.sync_from(source, flags=DiffSyncFlags.SKIP_UNMATCHED_DST)

# Only update objects that exist in both source and target
diff = target.sync_from(source, flags=DiffSyncFlags.SKIP_UNMATCHED_BOTH)

# Enable verbose logging for debugging
diff = target.sync_from(source, flags=DiffSyncFlags.LOG_UNCHANGED_RECORDS)

# Combine multiple flags using bitwise OR
diff = target.sync_from(source, 
                       flags=DiffSyncFlags.CONTINUE_ON_FAILURE | DiffSyncFlags.LOG_UNCHANGED_RECORDS)

DiffSyncModelFlags

Flags that can be set on a DiffSyncModel class or instance to affect its usage during diff and sync operations.

class DiffSyncModelFlags(enum.Flag):
    """Flags that can be set on a DiffSyncModel class or instance to affect its usage."""
    
    NONE = 0
    IGNORE = 0b1
    SKIP_CHILDREN_ON_DELETE = 0b10
    SKIP_UNMATCHED_SRC = 0b100
    SKIP_UNMATCHED_DST = 0b1000
    NATURAL_DELETION_ORDER = 0b10000
    SKIP_UNMATCHED_BOTH = SKIP_UNMATCHED_SRC | SKIP_UNMATCHED_DST

Model Flag Descriptions

NONE = 0

No special flags - default behavior.

IGNORE = 0b1

Do not render diffs containing this model; do not make any changes to this model when synchronizing. Can be used to indicate a model instance that exists but should not be changed by DiffSync.

SKIP_CHILDREN_ON_DELETE = 0b10

When deleting this model, do not recursively delete its children. Can be used for the case where deletion of a model results in the automatic deletion of all its children.

SKIP_UNMATCHED_SRC = 0b100

Ignore the model if it only exists in the source/"from" DiffSync when determining diffs and syncing. If this flag is set, no new model will be created in the target/"to" DiffSync.

SKIP_UNMATCHED_DST = 0b1000

Ignore the model if it only exists in the target/"to" DiffSync when determining diffs and syncing. If this flag is set, the model will not be deleted from the target/"to" DiffSync.

NATURAL_DELETION_ORDER = 0b10000

When deleting, delete children before instances of this element. If this flag is set, the model's children will be deleted from the target/"to" DiffSync before the model instances themselves.

SKIP_UNMATCHED_BOTH = SKIP_UNMATCHED_SRC | SKIP_UNMATCHED_DST

Combination flag - ignore the model if it exists only in source OR only in target.

DiffSyncModelFlags Usage Examples

from diffsync import DiffSyncModel, DiffSyncModelFlags

# Set flags as class attribute
class ReadOnlyDevice(DiffSyncModel):
    _modelname = "readonly_device"
    _identifiers = ("name",)
    _attributes = ("status",)
    
    # This model will be ignored during sync operations
    model_flags = DiffSyncModelFlags.IGNORE
    
    name: str
    status: str

# Set flags dynamically on instance
device = Device(name="critical_device", vendor="cisco")
device.model_flags = DiffSyncModelFlags.IGNORE

# Model that automatically deletes children when deleted
class Site(DiffSyncModel):
    _modelname = "site"
    _identifiers = ("name",)
    _children = {"device": "devices"}
    
    # When site is deleted, devices are automatically removed by the system
    model_flags = DiffSyncModelFlags.SKIP_CHILDREN_ON_DELETE
    
    name: str
    devices: List[str] = []

# Model with natural deletion order
class Database(DiffSyncModel):
    _modelname = "database"
    _identifiers = ("name",)
    _children = {"table": "tables"}
    
    # Delete tables before deleting database
    model_flags = DiffSyncModelFlags.NATURAL_DELETION_ORDER
    
    name: str
    tables: List[str] = []

# Conditional flags based on instance state
class Device(DiffSyncModel):
    _modelname = "device"
    _identifiers = ("name",)
    _attributes = ("status", "maintenance_mode")
    
    name: str
    status: str
    maintenance_mode: bool = False
    
    def __post_init__(self):
        # Don't sync devices in maintenance mode
        if self.maintenance_mode:
            self.model_flags = DiffSyncModelFlags.IGNORE

DiffSyncStatus

Enumeration of status values that can be set on a DiffSyncModel's status to indicate the result of create/update/delete operations.

class DiffSyncStatus(enum.Enum):
    """Flag values to set as a DiffSyncModel's _status when performing a sync; values are logged by DiffSyncSyncer."""
    
    UNKNOWN = "unknown"
    SUCCESS = "success"
    FAILURE = "failure"
    ERROR = "error"

Status Usage Examples

from diffsync import DiffSyncStatus

class NetworkDevice(DiffSyncModel):
    _modelname = "device"
    _identifiers = ("name",)
    _attributes = ("ip_address",)
    
    name: str
    ip_address: str
    
    @classmethod
    def create(cls, adapter, ids, attrs):
        device = super().create(adapter, ids, attrs)
        
        try:
            # Attempt to configure device
            result = configure_network_device(device.name, device.ip_address)
            if result.success:
                device.set_status(DiffSyncStatus.SUCCESS, "Device configured successfully")
            else:
                device.set_status(DiffSyncStatus.FAILURE, f"Configuration failed: {result.error}")
        except Exception as e:
            device.set_status(DiffSyncStatus.ERROR, f"Unexpected error: {e}")
        
        return device
    
    def update(self, attrs):
        device = super().update(attrs)
        
        if 'ip_address' in attrs:
            try:
                update_device_ip(self.name, self.ip_address)
                device.set_status(DiffSyncStatus.SUCCESS, "IP address updated")
            except ValidationError:
                device.set_status(DiffSyncStatus.FAILURE, "Invalid IP address")
            except NetworkError as e:
                device.set_status(DiffSyncStatus.ERROR, f"Network error: {e}")
        
        return device

# Check status after sync operations
diff = target.sync_from(source)
for element in diff.get_children():
    if element.action:
        try:
            obj = target.get(element.type, element.keys)
            status, message = obj.get_status()
            print(f"{element.type} {element.name}: {status.value} - {message}")
        except ObjectNotFound:
            print(f"{element.type} {element.name}: Object not found after sync")

Configuration Patterns

Environment-Based Configuration

import os
from diffsync import DiffSyncFlags

def get_sync_flags():
    """Get sync flags based on environment variables."""
    flags = DiffSyncFlags.NONE
    
    if os.getenv("DIFFSYNC_CONTINUE_ON_FAILURE", "false").lower() == "true":
        flags |= DiffSyncFlags.CONTINUE_ON_FAILURE
    
    if os.getenv("DIFFSYNC_SKIP_CREATES", "false").lower() == "true":
        flags |= DiffSyncFlags.SKIP_UNMATCHED_SRC
    
    if os.getenv("DIFFSYNC_SKIP_DELETES", "false").lower() == "true":
        flags |= DiffSyncFlags.SKIP_UNMATCHED_DST
    
    if os.getenv("DIFFSYNC_VERBOSE", "false").lower() == "true":
        flags |= DiffSyncFlags.LOG_UNCHANGED_RECORDS
    
    return flags

# Usage
flags = get_sync_flags()
diff = target.sync_from(source, flags=flags)

Configuration Classes

from dataclasses import dataclass
from diffsync import DiffSyncFlags, DiffSyncModelFlags

@dataclass
class SyncConfig:
    """Configuration for DiffSync operations."""
    continue_on_failure: bool = False
    skip_creates: bool = False
    skip_deletes: bool = False
    verbose_logging: bool = False
    dry_run: bool = False
    
    @property
    def flags(self) -> DiffSyncFlags:
        """Convert configuration to DiffSyncFlags."""
        flags = DiffSyncFlags.NONE
        
        if self.continue_on_failure:
            flags |= DiffSyncFlags.CONTINUE_ON_FAILURE
        if self.skip_creates:
            flags |= DiffSyncFlags.SKIP_UNMATCHED_SRC
        if self.skip_deletes:
            flags |= DiffSyncFlags.SKIP_UNMATCHED_DST
        if self.verbose_logging:
            flags |= DiffSyncFlags.LOG_UNCHANGED_RECORDS
        
        return flags

# Usage
config = SyncConfig(
    continue_on_failure=True,
    verbose_logging=True
)

if config.dry_run:
    # Just calculate diff without applying
    diff = target.diff_from(source)
    print("Dry run - would apply these changes:")
    print(diff.str())
else:
    # Apply changes
    diff = target.sync_from(source, flags=config.flags)

Conditional Flag Application

def smart_sync(source_adapter, target_adapter, safety_mode=True):
    """Perform sync with intelligent flag selection based on conditions."""
    
    # Start with basic flags
    flags = DiffSyncFlags.NONE
    
    # In safety mode, be more conservative
    if safety_mode:
        flags |= DiffSyncFlags.CONTINUE_ON_FAILURE
        flags |= DiffSyncFlags.LOG_UNCHANGED_RECORDS
    
    # Calculate diff first to analyze changes
    diff = target_adapter.diff_from(source_adapter)
    summary = diff.summary()
    
    # If there are many deletions, ask for confirmation
    if summary['delete'] > 10:
        print(f"Warning: {summary['delete']} objects will be deleted")
        if input("Continue? (y/n): ").lower() != 'y':
            flags |= DiffSyncFlags.SKIP_UNMATCHED_DST
    
    # If there are many creates, be cautious about resources
    if summary['create'] > 100:
        print(f"Warning: {summary['create']} objects will be created")
        # Could add resource checks here
    
    # Perform sync with selected flags
    return target_adapter.sync_from(source_adapter, flags=flags, diff=diff)

Logging Configuration

DiffSync uses structured logging through the structlog library. Configure logging verbosity and format using the logging utilities.

def enable_console_logging(verbosity: int = 0) -> None:
    """
    Enable formatted logging to console with the specified verbosity.
    
    Args:
        verbosity: 0 for WARNING logs, 1 for INFO logs, 2 for DEBUG logs
    """

Logging Usage Examples

from diffsync.logging import enable_console_logging

# Enable INFO level logging
enable_console_logging(verbosity=1)

# Enable DEBUG level logging for detailed troubleshooting
enable_console_logging(verbosity=2)

# Perform sync with verbose logging
diff = target.sync_from(source, flags=DiffSyncFlags.LOG_UNCHANGED_RECORDS)

DiffSyncActions

Constants representing valid actions for DiffSyncModel operations during synchronization.

class DiffSyncActions:
    """List of valid Action for DiffSyncModel."""
    
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"  
    SKIP = "skip"
    NO_CHANGE = None

These constants are used internally by DiffSync to represent the different types of operations that can be performed on models during synchronization. Import from the enum module:

from diffsync.enum import DiffSyncActions

# Usage example - these values appear in DiffElement.action property
for element in diff.get_children():
    if element.action == DiffSyncActions.CREATE:
        print(f"Will create {element.name}")
    elif element.action == DiffSyncActions.UPDATE:
        print(f"Will update {element.name}")
    elif element.action == DiffSyncActions.DELETE:
        print(f"Will delete {element.name}")

Types

import enum
from typing import Union

# Flag types for type hints
SyncFlags = Union[DiffSyncFlags, int]
ModelFlags = Union[DiffSyncModelFlags, int]

# Status enumeration
StatusValue = DiffSyncStatus

Install with Tessl CLI