database

django_spire.contrib.sync.database

__all__ = ['ConflictEntry', 'ConflictResolver', 'ConflictType', 'DatabaseEngine', 'DatabaseResult', 'DatabaseSyncStorage', 'DependencyGraph', 'FieldConflict', 'FieldOwnershipWins', 'FieldTimestampWins', 'FieldUpdateTracker', 'LocalWins', 'ModelPayload', 'PayloadReconciler', 'ReconciliationResult', 'RecordConflict', 'RecordResolution', 'RemoteWins', 'ResolutionSource', 'SyncLock', 'SyncManifest', 'SyncRecord'] module-attribute

ConflictResolver

Bases: Protocol

resolve

Source code in django_spire/contrib/sync/database/conflict.py

def resolve(
    self,
    conflict: RecordConflict,
) -> RecordResolution: ...

ConflictType

Bases: StrEnum

BOTH_MODIFIED = 'both_modified' class-attribute instance-attribute

COMPATIBLE = 'compatible' class-attribute instance-attribute

DELETE_VS_MODIFY = 'delete_vs_modify' class-attribute instance-attribute

MODIFY_VS_DELETE = 'modify_vs_delete' class-attribute instance-attribute

FieldConflict dataclass

field_name instance-attribute

local_value instance-attribute

remote_value instance-attribute

local_timestamp instance-attribute

remote_timestamp instance-attribute

FieldOwnershipWins

Source code in django_spire/contrib/sync/database/conflict.py

def __init__(
    self,
    local_fields: set[str],
    remote_fields: set[str],
    exclude_fields: set[str] | None = None,
    prefer_remote_on_tie: bool = False,
) -> None:
    overlap = local_fields & remote_fields

    if overlap:
        message = (
            f'local_fields and remote_fields must not '
            f'overlap: {overlap}'
        )

        raise InvalidParameterError(message)

    self._exclude = frozenset(exclude_fields or set()) | META_FIELDS
    self._local_fields = frozenset(local_fields)
    self._prefer_remote_on_tie = prefer_remote_on_tie
    self._remote_fields = frozenset(remote_fields)

resolve

Source code in django_spire/contrib/sync/database/conflict.py

def resolve(
    self,
    conflict: RecordConflict,
) -> RecordResolution:
    if conflict.conflict_type == ConflictType.DELETE_VS_MODIFY:
        return RecordResolution(
            record=_require_local(conflict),
            source=ResolutionSource.LOCAL,
        )

    if conflict.conflict_type == ConflictType.MODIFY_VS_DELETE:
        return RecordResolution(
            record=_require_remote(conflict),
            source=ResolutionSource.REMOTE,
        )

    local, remote = _require_both(conflict)

    return _merge_fields(
        local,
        remote,
        conflict,
        self._exclude,
        self._prefer_remote_on_tie,
        local_fields=self._local_fields,
        remote_fields=self._remote_fields,
    )

FieldTimestampWins

Source code in django_spire/contrib/sync/database/conflict.py

def __init__(
    self,
    exclude_fields: set[str] | None = None,
    prefer_remote_on_tie: bool = False,
) -> None:
    self._exclude = frozenset(exclude_fields or set()) | META_FIELDS
    self._prefer_remote_on_tie = prefer_remote_on_tie

resolve

Source code in django_spire/contrib/sync/database/conflict.py

def resolve(
    self,
    conflict: RecordConflict,
) -> RecordResolution:
    if conflict.conflict_type == ConflictType.DELETE_VS_MODIFY:
        return RecordResolution(
            record=_require_local(conflict),
            source=ResolutionSource.LOCAL,
        )

    if conflict.conflict_type == ConflictType.MODIFY_VS_DELETE:
        return RecordResolution(
            record=_require_remote(conflict),
            source=ResolutionSource.REMOTE,
        )

    local, remote = _require_both(conflict)

    return _merge_fields(
        local,
        remote,
        conflict,
        self._exclude,
        self._prefer_remote_on_tie,
    )

LocalWins

resolve

Source code in django_spire/contrib/sync/database/conflict.py

def resolve(
    self,
    conflict: RecordConflict,
) -> RecordResolution:
    if conflict.conflict_type == ConflictType.MODIFY_VS_DELETE:
        return RecordResolution(
            record=None,
            source=ResolutionSource.LOCAL,
            delete=True,
        )

    return RecordResolution(
        record=_require_local(conflict),
        source=ResolutionSource.LOCAL,
        field_conflicts=conflict.field_conflicts,
    )

RecordConflict dataclass

key instance-attribute

model_label instance-attribute

conflict_type instance-attribute

field_conflicts = field(default_factory=list) class-attribute instance-attribute

local = None class-attribute instance-attribute

remote = None class-attribute instance-attribute

RecordResolution dataclass

record instance-attribute

source instance-attribute

delete = False class-attribute instance-attribute

field_conflicts = field(default_factory=list) class-attribute instance-attribute

RemoteWins

resolve

Source code in django_spire/contrib/sync/database/conflict.py

def resolve(
    self,
    conflict: RecordConflict,
) -> RecordResolution:
    if conflict.conflict_type == ConflictType.DELETE_VS_MODIFY:
        return RecordResolution(
            record=None,
            source=ResolutionSource.REMOTE,
            delete=True,
        )

    return RecordResolution(
        record=_require_remote(conflict),
        source=ResolutionSource.REMOTE,
        field_conflicts=conflict.field_conflicts,
    )

ResolutionSource

Bases: StrEnum

LOCAL = 'local' class-attribute instance-attribute

MERGED = 'merged' class-attribute instance-attribute

REMOTE = 'remote' class-attribute instance-attribute

DatabaseEngine

Source code in django_spire/contrib/sync/database/engine.py

def __init__(
    self,
    storage: DatabaseSyncStorage,
    graph: DependencyGraph,
    clock: HybridLogicalClock,
    node_id: str,
    *,
    batch_bytes: int | None = BATCH_BYTES_DEFAULT,
    batch_size: int | None = None,
    clock_drift_max: int | None = CLOCK_DRIFT_MAX_DEFAULT,
    foreign_key_columns: dict[str, list[tuple[str, str]]] | None = None,
    identity_field: str = 'id',
    lock: SyncLock | None = None,
    on_complete: Callable[[DatabaseResult], None] | None = None,
    on_phase: Callable[[SyncPhase], None] | None = None,
    payload_bytes_max: int | None = None,
    payload_records_max: int | None = None,
    peer_node_id: str | None = None,
    progress: Callable[[SyncStage, int, int], None] | None = None,
    reconciler: PayloadReconciler | None = None,
    transaction: Callable[[], AbstractContextManager[Any]] = nullcontext,
    transport: Transport | None = None,
) -> None:
    if not node_id:
        message = 'node_id must be a non-empty string'
        raise InvalidParameterError(message)

    if not identity_field:
        message = 'identity_field must be a non-empty string'
        raise InvalidParameterError(message)

    if batch_bytes is not None and batch_bytes < 1:
        message = (
            f'batch_bytes must be >= 1 '
            f'or None, got {batch_bytes}'
        )

        raise InvalidParameterError(message)

    if batch_size is not None and batch_size < 1:
        message = (
            f'batch_size must be >= 1 '
            f'or None, got {batch_size}'
        )

        raise InvalidParameterError(message)

    if clock_drift_max is not None and clock_drift_max < 0:
        message = (
            f'clock_drift_max must be non-negative '
            f'or None, got {clock_drift_max}'
        )

        raise InvalidParameterError(message)

    if payload_bytes_max is not None and payload_bytes_max < 1:
        message = (
            f'payload_bytes_max must be >= 1 '
            f'or None, got {payload_bytes_max}'
        )

        raise InvalidParameterError(message)

    if payload_records_max is not None and payload_records_max < 1:
        message = (
            f'payload_records_max must be >= 1 '
            f'or None, got {payload_records_max}'
        )

        raise InvalidParameterError(message)

    if transport is not None and not peer_node_id:
        message = (
            'peer_node_id is required when transport is set '
            '(client mode)'
        )

        raise InvalidParameterError(message)

    if peer_node_id is not None and peer_node_id == node_id:
        message = (
            f'peer_node_id must differ from node_id '
            f'(both are {node_id!r})'
        )

        raise InvalidParameterError(message)

    self._batch_bytes = batch_bytes
    self._batch_size = batch_size
    self._clock = clock
    self._clock_drift_max = clock_drift_max
    self._errored_keys: dict[str, set[str]] = defaultdict(set)
    self._foreign_key_columns = foreign_key_columns or {}
    self._graph = graph
    self._identity_field = identity_field
    self._lock = lock
    self._node_id = node_id
    self._on_complete = on_complete
    self._on_phase = on_phase
    self._payload_bytes_max = payload_bytes_max
    self._payload_records_max = payload_records_max
    self._peer_node_id = peer_node_id or ''
    self._progress = progress
    self._reconciler = reconciler or PayloadReconciler()
    self._storage = storage
    self._transaction = transaction
    self._transport = transport

process

Source code in django_spire/contrib/sync/database/engine.py

def process(
    self,
    incoming: SyncManifest,
) -> tuple[SyncManifest, DatabaseResult]:
    result = DatabaseResult()

    self._reset_errored_keys()
    self._stamp_unstamped_records()

    self._validate_manifest(incoming)
    self._validate_clock(incoming)

    valid_payloads = self._validate_incoming_models(
        incoming,
        result,
    )

    peer_node_id = incoming.node_id

    with self._transaction():
        if self._lock:
            self._lock.hold_global()
            self._lock.hold(self._node_id, peer_node_id)

        counter_at_start = (
            self._storage.get_sequence_allocator().current()
        )

        response_payloads, has_more, after_keys, response_sequence_max = (
            self._apply_incoming(
                valid_payloads,
                incoming.peer_sequence,
                peer_node_id,
                result,
                sequence_max=counter_at_start,
                records_max=self._batch_size,
                bytes_max=self._batch_bytes,
                after_keys=incoming.after_keys,
            )
        )

        self._flush_deferred_backfill()

    self._advance_clock(incoming)

    if has_more:
        outgoing_local_sequence = response_sequence_max
    else:
        outgoing_local_sequence = counter_at_start

    response = SyncManifest(
        node_id=self._node_id,
        peer_sequence=incoming.local_sequence,
        local_sequence=outgoing_local_sequence,
        after_keys=after_keys,
        node_time=int(time.time()),
        payloads=response_payloads,
        has_more=has_more,
    )

    response.checksum = response.compute_checksum()

    self._finalize(result)

    return response, result

sync

Source code in django_spire/contrib/sync/database/engine.py

def sync(self, dry_run: bool = False) -> DatabaseResult:
    if self._transport is None:
        message = (
            'Transport is required for sync(). '
            'Use process() for server-side.'
        )

        raise TransportRequiredError(message)

    if not self._peer_node_id:
        message = (
            'peer_node_id is required for sync(). '
            'Configure it on the engine.'
        )

        raise InvalidParameterError(message)

    result = DatabaseResult()

    self._reset_errored_keys()

    persisted = self._storage.get_after_keys(self._peer_node_id)

    server_cursors: dict[str, Any] = {
        key.removeprefix('server:'): value
        for key, value in persisted.items()
        if key.startswith('server:')
    }

    collect_cursors: dict[str, Any] = {
        key.removeprefix('collect:'): value
        for key, value in persisted.items()
        if key.startswith('collect:')
    }

    iteration = 0

    with self._managed_session(result) as session_id:
        self._stamp_unstamped_records()

        while True:
            iteration += 1

            self._enter_phase(
                SyncPhase.COLLECTING,
                session_id,
                SyncStage.VALIDATE,
            )

            checkpoint = self._storage.get_checkpoint(self._peer_node_id)
            peer_sequence = checkpoint.peer_sequence
            local_sequence_pushed = checkpoint.local_sequence_pushed

            counter_at_start = (
                self._storage.get_sequence_allocator().current()
            )

            manifest = self._collect(
                local_sequence_pushed,
                self._peer_node_id,
                sequence_max=counter_at_start,
                limit=self._batch_size,
                bytes_limit=self._batch_bytes,
                after_keys=collect_cursors,
            )

            collected_cursors = manifest.after_keys or {}

            manifest.peer_sequence = peer_sequence
            manifest.after_keys = server_cursors
            manifest.checksum = manifest.compute_checksum()

            sent_max_sequence = manifest.local_sequence
            sent_has_more = manifest.has_more

            self._record_pushed(manifest, result)

            self._enter_phase(
                SyncPhase.EXCHANGING,
                session_id,
                SyncStage.CLASSIFY,
            )

            response = self._exchange_and_validate(manifest)

            self._enter_phase(
                SyncPhase.RECONCILING,
                session_id,
                SyncStage.MUTATE,
            )

            if manifest.has_more:
                collect_cursors = collected_cursors
            else:
                collect_cursors = {}

            if response.has_more:
                server_cursors = response.after_keys
            else:
                server_cursors = {}

            if not dry_run:
                self._enter_phase(
                    SyncPhase.COMMITTING,
                    session_id,
                )

                self._commit(
                    peer_sequence,
                    local_sequence_pushed,
                    sent_max_sequence,
                    sent_has_more,
                    counter_at_start,
                    response,
                    result,
                    server_cursors=server_cursors,
                    collect_cursors=collect_cursors,
                )

            self._advance_clock(response)

            converged = (
                not manifest.has_more
                and not response.has_more
            )

            if dry_run:
                break

            if converged:
                exchanged = (
                    any(payload.records or payload.deletes for payload in manifest.payloads)
                    or any(payload.records or payload.deletes for payload in response.payloads)
                )

                if not exchanged:
                    break

        self._enter_phase(SyncPhase.COMPLETE, session_id)
        self._finalize(result)

    # self._log_sync_summary(result)

    return result

DependencyGraph

Source code in django_spire/contrib/sync/core/graph.py

def __init__(
    self,
    edges: dict[str, set[str]],
    deferred_edges: dict[str, set[str]] | None = None,
) -> None:
    for label in edges:
        if not label:
            message = 'edges must not contain empty labels'
            raise InvalidParameterError(message)

    self._edges = {
        label: set(dependencies)
        for label, dependencies in edges.items()
    }

    all_labels = set(self._edges)

    for label, dependencies in self._edges.items():
        unknown = dependencies - all_labels

        if unknown:
            message = (
                f'Model {label!r} declares dependencies on '
                f'unknown models: {unknown}'
            )

            raise UnknownDependencyError(message)

    self._deferred_edges: dict[str, set[str]] = {}

    if deferred_edges:
        for label, targets in deferred_edges.items():
            if not targets:
                continue

            if label not in all_labels:
                message = (
                    f'Deferred edge source {label!r} '
                    f'is not a known model'
                )

                raise InvalidParameterError(message)

            unknown = targets - all_labels

            if unknown:
                message = (
                    f'Deferred edges from {label!r} reference '
                    f'unknown models: {unknown}'
                )

                raise UnknownDependencyError(message)

            self._deferred_edges[label] = set(targets)

    self._dependents: dict[str, set[str]] = {
        label: set() for label in self._edges
    }

    for label, dependencies in self._edges.items():
        for dependency in dependencies:
            self._dependents[dependency].add(label)

    self._order = self._compute_order()

deferred_edges property

dependencies

Source code in django_spire/contrib/sync/core/graph.py

def dependencies(self, label: str) -> set[str]:
    return set(self._edges.get(label, set()))

has_deferred_edges

Source code in django_spire/contrib/sync/core/graph.py

def has_deferred_edges(self) -> bool:
    return bool(self._deferred_edges)

known_models

Source code in django_spire/contrib/sync/core/graph.py

def known_models(self) -> frozenset[str]:
    return frozenset(self._edges)

sync_order

Source code in django_spire/contrib/sync/core/graph.py

def sync_order(self) -> list[str]:
    return list(self._order)

SyncLock

Bases: Protocol

acquire

Source code in django_spire/contrib/sync/database/lock.py

def acquire(self, node_id: str, peer_node_id: str) -> str: ...

hold

Source code in django_spire/contrib/sync/database/lock.py

def hold(self, node_id: str, peer_node_id: str) -> None: ...

hold_global

Source code in django_spire/contrib/sync/database/lock.py

def hold_global(self) -> None: ...

release

Source code in django_spire/contrib/sync/database/lock.py

def release(self, session_id: str, status: SyncStatus, result: DatabaseResult | None = None) -> None: ...

update_phase

Source code in django_spire/contrib/sync/database/lock.py

def update_phase(self, session_id: str, phase: SyncPhase) -> None: ...

ConflictEntry dataclass

conflict instance-attribute

resolution_source instance-attribute

DatabaseResult dataclass

applied = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

compatible = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

conflict_log = field(default_factory=list) class-attribute instance-attribute

conflicts = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

created = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

deleted = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

errors = field(default_factory=list) class-attribute instance-attribute

pushed = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

skipped = field(default_factory=(lambda: defaultdict(list))) class-attribute instance-attribute

ok property

ModelPayload dataclass

model_label instance-attribute

records = field(default_factory=dict) class-attribute instance-attribute

deletes = field(default_factory=dict) class-attribute instance-attribute

from_dict classmethod

Source code in django_spire/contrib/sync/database/manifest.py

@classmethod
def from_dict(
    cls,
    data: dict[str, Any],
) -> ModelPayload:
    model_label = data.get('model_label')

    if model_label is None:
        message = "ModelPayload requires 'model_label'"
        raise ManifestFieldError(message)

    if not isinstance(model_label, str):
        message = "'model_label' must be a string"
        raise ManifestFieldError(message)

    if not model_label:
        message = "'model_label' must be a non-empty string"
        raise ManifestFieldError(message)

    raw_records = data.get('records', {})

    if not isinstance(raw_records, dict):
        message = "'records' must be a dict"
        raise ManifestFieldError(message)

    raw_deletes = data.get('deletes', {})

    if not isinstance(raw_deletes, dict):
        message = "'deletes' must be a dict"
        raise ManifestFieldError(message)

    deletes: dict[str, int] = {}

    for key, tombstone in raw_deletes.items():
        if not isinstance(key, str):
            message = f'delete key {key!r} must be a string'
            raise ManifestFieldError(message)

        if (
            not isinstance(tombstone, int)
            or isinstance(tombstone, bool)
        ):
            message = (
                f"delete tombstone for {key!r} must be "
                f"an integer, got "
                f"{type(tombstone).__name__}"
            )

            raise ManifestFieldError(message)

        if tombstone < 0:
            message = (
                f"delete tombstone for {key!r} must be "
                f"non-negative, got {tombstone}"
            )

            raise ManifestFieldError(message)

        if key in raw_records:
            message = (
                f"key {key!r} present in both "
                f"'records' and 'deletes'"
            )

            raise ManifestFieldError(message)

        deletes[key] = tombstone

    return cls(
        model_label=model_label,
        records={
            key: SyncRecord.from_dict(key, value)
            for key, value in raw_records.items()
        },
        deletes=deletes,
    )

to_dict

Source code in django_spire/contrib/sync/database/manifest.py

def to_dict(self) -> dict[str, Any]:
    return {
        'deletes': dict(sorted(self.deletes.items())),
        'model_label': self.model_label,
        'records': {
            key: record.to_dict()
            for key, record in self.records.items()
        },
    }

SyncManifest dataclass

node_id instance-attribute

peer_sequence = 0 class-attribute instance-attribute

local_sequence = 0 class-attribute instance-attribute

after_keys = field(default_factory=dict) class-attribute instance-attribute

checksum = '' class-attribute instance-attribute

has_more = False class-attribute instance-attribute

node_time = 0 class-attribute instance-attribute

payloads = field(default_factory=list) class-attribute instance-attribute

compute_checksum

Source code in django_spire/contrib/sync/database/manifest.py

def compute_checksum(self) -> str:
    body = json.dumps(
        self._serializable(),
        sort_keys=True,
        ensure_ascii=True,
    ).encode('utf-8')

    return hashlib.sha256(body).hexdigest()

verify

Source code in django_spire/contrib/sync/database/manifest.py

def verify(self) -> bool:
    if not self.checksum:
        return False

    return self.checksum == self.compute_checksum()

from_dict classmethod

Source code in django_spire/contrib/sync/database/manifest.py

@classmethod
def from_dict(
    cls,
    data: dict[str, Any],
) -> SyncManifest:
    node_id = data.get('node_id')
    peer_sequence = data.get('peer_sequence')
    local_sequence = data.get('local_sequence')

    if node_id is None:
        message = "SyncManifest requires 'node_id'"
        raise ManifestFieldError(message)

    if not isinstance(node_id, str):
        message = "'node_id' must be a string"
        raise ManifestFieldError(message)

    if not node_id:
        message = "'node_id' must be a non-empty string"
        raise ManifestFieldError(message)

    if peer_sequence is None:
        message = "SyncManifest requires 'peer_sequence'"
        raise ManifestFieldError(message)

    if (
        not isinstance(peer_sequence, int)
        or isinstance(peer_sequence, bool)
    ):
        message = "'peer_sequence' must be an integer"
        raise ManifestFieldError(message)

    if peer_sequence < 0:
        message = (
            f"'peer_sequence' must be non-negative, "
            f"got {peer_sequence}"
        )

        raise ManifestFieldError(message)

    if local_sequence is None:
        message = "SyncManifest requires 'local_sequence'"
        raise ManifestFieldError(message)

    if (
        not isinstance(local_sequence, int)
        or isinstance(local_sequence, bool)
    ):
        message = "'local_sequence' must be an integer"
        raise ManifestFieldError(message)

    if local_sequence < 0:
        message = (
            f"'local_sequence' must be non-negative, "
            f"got {local_sequence}"
        )

        raise ManifestFieldError(message)

    node_time = data.get('node_time', 0)

    if (
        not isinstance(node_time, int)
        or isinstance(node_time, bool)
    ):
        message = "'node_time' must be an integer"
        raise ManifestFieldError(message)

    if node_time < 0:
        message = (
            f"'node_time' must be non-negative, "
            f"got {node_time}"
        )

        raise ManifestFieldError(message)

    after_keys = data.get('after_keys', {})
    has_more = data.get('has_more', False)

    raw_payloads = data.get('payloads', [])

    if not isinstance(raw_payloads, list):
        message = "'payloads' must be a list"
        raise ManifestFieldError(message)

    if len(raw_payloads) > _PAYLOADS_MAX:
        message = (
            f"'payloads' exceeds maximum of "
            f"{_PAYLOADS_MAX}"
        )

        raise ManifestFieldError(message)

    seen_labels: set[str] = set()
    payloads: list[ModelPayload] = []

    for raw_payload in raw_payloads:
        payload = ModelPayload.from_dict(raw_payload)

        if payload.model_label in seen_labels:
            message = (
                f"duplicate model_label: "
                f"{payload.model_label!r}"
            )

            raise ManifestFieldError(message)

        seen_labels.add(payload.model_label)
        payloads.append(payload)

    return cls(
        node_id=node_id,
        peer_sequence=peer_sequence,
        local_sequence=local_sequence,
        after_keys=after_keys if isinstance(after_keys, dict) else {},
        checksum=data.get('checksum', ''),
        has_more=has_more,
        node_time=node_time,
        payloads=payloads,
    )

to_dict

Source code in django_spire/contrib/sync/database/manifest.py

def to_dict(self) -> dict[str, Any]:
    result = self._serializable()
    result['after_keys'] = self.after_keys
    result['checksum'] = self.checksum or self.compute_checksum()
    result['has_more'] = self.has_more

    return result

PayloadReconciler

Source code in django_spire/contrib/sync/database/reconciler.py

def __init__(
    self,
    resolver: ConflictResolver | None = None,
) -> None:
    self._resolver = resolver or FieldTimestampWins()

reconcile

Source code in django_spire/contrib/sync/database/reconciler.py

def reconcile(
    self,
    payload: ModelPayload,
    local_records: dict[str, SyncRecord],
    local_tombstones: dict[str, int],
) -> ReconciliationResult:
    result = ReconciliationResult()

    for key, remote in payload.records.items():
        self._classify_record(
            key,
            remote,
            payload.model_label,
            local_records,
            local_tombstones,
            result,
        )

    self._classify_deletes(payload, local_records, result)

    return result

ReconciliationResult dataclass

compatible_keys = field(default_factory=list) class-attribute instance-attribute

conflict_keys = field(default_factory=list) class-attribute instance-attribute

conflict_log = field(default_factory=list) class-attribute instance-attribute

created_keys = field(default_factory=set) class-attribute instance-attribute

errors = field(default_factory=list) class-attribute instance-attribute

response_records = field(default_factory=dict) class-attribute instance-attribute

to_clear_tombstones = field(default_factory=set) class-attribute instance-attribute

to_delete = field(default_factory=dict) class-attribute instance-attribute

to_upsert = field(default_factory=dict) class-attribute instance-attribute

SyncRecord dataclass

key instance-attribute

data instance-attribute

timestamps instance-attribute

sequence = field(default=0, compare=False) class-attribute instance-attribute

origin_node = field(default='', compare=False) class-attribute instance-attribute

received_at = field(default=0, compare=False) class-attribute instance-attribute

sync_field_last_modified property

__post_init__

Source code in django_spire/contrib/sync/database/record.py

def __post_init__(self) -> None:
    if self.key == '':
        message = 'SyncRecord key must be a non-empty string'
        raise InvalidParameterError(message)

    if self.received_at < 0:
        message = (
            f'received_at must be non-negative, '
            f'got {self.received_at}'
        )

        raise InvalidParameterError(message)

    if self.sequence < 0:
        message = (
            f'sequence must be non-negative, '
            f'got {self.sequence}'
        )

        raise InvalidParameterError(message)

from_dict classmethod

Source code in django_spire/contrib/sync/database/record.py

@classmethod
def from_dict(cls, key: str, data: dict[str, Any]) -> SyncRecord:
    if key == '':
        message = 'SyncRecord key must be a non-empty string'
        raise RecordFieldError(message)

    record_data = data.get('data', {})
    record_timestamps = data.get('timestamps', {})
    record_received_at = data.get('received_at', 0)
    record_sequence = data.get('sequence', 0)
    record_origin_node = data.get('origin_node', '')

    if not isinstance(record_data, dict):
        message = (
            f"Record {key!r}: 'data' must be a dict, "
            f'got {type(record_data).__name__}'
        )

        raise RecordFieldError(message)

    if not isinstance(record_timestamps, dict):
        message = (
            f"Record {key!r}: 'timestamps' must be a "
            f'dict, got {type(record_timestamps).__name__}'
        )

        raise RecordFieldError(message)

    record_received_at = _coerce_int(
        record_received_at,
        "'received_at'",
        key,
    )

    if record_received_at < 0:
        message = (
            f"Record {key!r}: 'received_at' must be "
            f'non-negative, got {record_received_at}'
        )

        raise RecordFieldError(message)

    record_sequence = _coerce_int(
        record_sequence,
        "'sequence'",
        key,
    )

    if record_sequence < 0:
        message = (
            f"Record {key!r}: 'sequence' must be "
            f'non-negative, got {record_sequence}'
        )

        raise RecordFieldError(message)

    if not isinstance(record_origin_node, str):
        message = (
            f"Record {key!r}: 'origin_node' must be a string, "
            f'got {type(record_origin_node).__name__}'
        )

        raise RecordFieldError(message)

    sanitized_timestamps: dict[str, int] = {}

    for timestamp_key, timestamp_value in record_timestamps.items():
        if not isinstance(timestamp_key, str):
            message = (
                f'Record {key!r}: timestamp key '
                f'{timestamp_key!r} must be a string'
            )

            raise RecordFieldError(message)

        coerced = _coerce_int(
            timestamp_value,
            f'timestamp for {timestamp_key!r}',
            key,
        )

        if coerced < 0:
            message = (
                f'Record {key!r}: timestamp for '
                f'{timestamp_key!r} must be non-negative, '
                f'got {coerced}'
            )

            raise RecordFieldError(message)

        sanitized_timestamps[timestamp_key] = coerced

    return cls(
        key=key,
        data=record_data,
        timestamps=sanitized_timestamps,
        sequence=record_sequence,
        origin_node=record_origin_node,
        received_at=record_received_at,
    )

to_dict

Source code in django_spire/contrib/sync/database/record.py

def to_dict(self) -> dict[str, Any]:
    return {
        'data': self.data,
        'origin_node': self.origin_node,
        'sequence': self.sequence,
        'timestamps': self.timestamps,
    }

DatabaseSyncStorage

Bases: CheckpointStore, RecordReader, RecordWriter, Protocol

get_sequence_allocator

Source code in django_spire/contrib/sync/database/storage.py

def get_sequence_allocator(self) -> SequenceAllocator: ...

FieldUpdateTracker

Source code in django_spire/contrib/sync/database/tracker.py

def __init__(self) -> None:
    self._original: dict[str, Any] = {}

get_dirty

Source code in django_spire/contrib/sync/database/tracker.py

def get_dirty(self, current: dict[str, Any]) -> set[str]:
    dirty: set[str] = set()

    for key, value in current.items():
        if key not in self._original or self._original[key] != value:
            dirty.add(key)

    return dirty

snapshot

Source code in django_spire/contrib/sync/database/tracker.py

def snapshot(self, fields: dict[str, Any]) -> None:
    self._original = {
        key: copy.deepcopy(value) if isinstance(value, _MUTABLE_TYPES) else value
        for key, value in fields.items()
    }