conflict

django_spire.contrib.sync.database.conflict

META_FIELDS = frozenset({'sync_field_timestamps', 'sync_field_last_modified'}) module-attribute

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

ResolutionSource

Bases: StrEnum

LOCAL = 'local' class-attribute instance-attribute

MERGED = 'merged' class-attribute instance-attribute

REMOTE = 'remote' 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

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

ConflictResolver

Bases: Protocol

resolve

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

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

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,
    )

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,
    )