manifest

django_spire.contrib.sync.database.manifest

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

checkpoint 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')
    checkpoint = data.get('checkpoint')

    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 checkpoint is None:
        message = "SyncManifest requires 'checkpoint'"
        raise ManifestFieldError(message)

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

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

        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,
        checkpoint=checkpoint,
        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

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