Skip to content

mixin

django_spire.contrib.sync.django.mixin

SyncableFieldsMixin

Bases: Model

Source code in django_spire/contrib/sync/django/mixin.py 
61
62
63
64
65
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)

    self._tracker = FieldUpdateTracker()
    self._tracker.snapshot(self._get_field_values())

sync_field_timestamps = models.JSONField(default=dict, editable=False) class-attribute instance-attribute

sync_field_last_modified = models.BigIntegerField(default=0, editable=False, db_index=True) class-attribute instance-attribute

sync_field_sequence = models.BigIntegerField(default=0, editable=False, db_index=True) class-attribute instance-attribute

sync_field_origin_node = models.CharField(max_length=255, default='', editable=False, blank=True) class-attribute instance-attribute

objects = SyncableQuerySet.as_manager() class-attribute instance-attribute

Meta

abstract = True class-attribute instance-attribute

delete

Source code in django_spire/contrib/sync/django/mixin.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def delete(self, *args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]:
    if _is_bypassed():
        return super().delete(*args, **kwargs)

    if not hasattr(self, 'is_deleted'):
        return super().delete(*args, **kwargs)

    if self.is_deleted:
        return 0, {}

    self.is_deleted = True
    self.save()

    return 1, {self._meta.label: 1}

save

Source code in django_spire/contrib/sync/django/mixin.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def save(self, *args: Any, **kwargs: Any) -> None:
    if _is_bypassed():
        super().save(*args, **kwargs)
        self._tracker.snapshot(self._get_field_values())

        return

    dirty = self.get_dirty_fields()

    if not dirty:
        super().save(*args, **kwargs)
        self._tracker.snapshot(self._get_field_values())

        return

    from django_spire.contrib.sync.django.sequence import (  # noqa: PLC0415
        SyncSequenceAllocator,
    )

    with transaction.atomic(using=self._state.db):
        now = self.get_clock().now()
        timestamps = dict(self.sync_field_timestamps)

        for field_name in dirty:
            timestamps[field_name] = now

        sequence_last = SyncSequenceAllocator(using=self._state.db).allocate(1).value_last

        self.sync_field_timestamps = timestamps
        self.sync_field_last_modified = now
        self.sync_field_sequence = sequence_last
        self.sync_field_origin_node = ''

        super().save(*args, **kwargs)
        self._tracker.snapshot(self._get_field_values())

get_dirty_fields

Source code in django_spire/contrib/sync/django/mixin.py 
125
126
127
128
129
def get_dirty_fields(self) -> set[str]:
    if self._state.adding:
        return set(self.get_syncable_field_names())

    return self._tracker.get_dirty(self._get_field_values())

refresh_from_db

Source code in django_spire/contrib/sync/django/mixin.py 
131
132
133
134
135
136
137
def refresh_from_db(
    self,
    *args: Any,
    **kwargs: Any,
) -> None:
    super().refresh_from_db(*args, **kwargs)
    self._tracker.snapshot(self._get_field_values())

configure classmethod

Source code in django_spire/contrib/sync/django/mixin.py 
139
140
141
@classmethod
def configure(cls, clock: HybridLogicalClock) -> None:
    cls._clock = clock

get_clock classmethod

Source code in django_spire/contrib/sync/django/mixin.py 
143
144
145
146
147
148
149
150
151
152
153
154
@classmethod
def get_clock(cls) -> HybridLogicalClock:
    if cls._clock is None:
        message = (
            'SyncableMixin clock not configured. '
            'Call SyncableMixin.configure(clock) '
            'in AppConfig.ready().'
        )

        raise ClockNotConfiguredError(message)

    return cls._clock

get_syncable_field_names classmethod

Source code in django_spire/contrib/sync/django/mixin.py 
156
157
158
159
160
161
162
@classmethod
def get_syncable_field_names(cls) -> list[str]:
    return sorted(
        field.attname if field.is_relation else field.name
        for field in cls._meta.concrete_fields
        if field.name not in cls._sync_exclude_fields
    )

get_syncable_many_to_many_names classmethod

Source code in django_spire/contrib/sync/django/mixin.py 
164
165
166
167
168
169
@classmethod
def get_syncable_many_to_many_names(cls) -> list[str]:
    return sorted(
        field.name
        for field in cls._meta.many_to_many
    )

SyncableMixin

Bases: SyncableFieldsMixin

Source code in django_spire/contrib/sync/django/mixin.py 
61
62
63
64
65
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)

    self._tracker = FieldUpdateTracker()
    self._tracker.snapshot(self._get_field_values())

id = models.UUIDField(primary_key=True, default=(uuid.uuid4), editable=False) class-attribute instance-attribute

Meta

abstract = True class-attribute instance-attribute