Skip to content

queryset

django_spire.contrib.sync.django.queryset

SyncableQuerySet

Bases: QuerySet

bulk_create

Source code in django_spire/contrib/sync/django/queryset.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def bulk_create(
    self,
    objs: list[Any],
    **kwargs: Any,
) -> list[Any]:
    if _is_bypassed():
        return super().bulk_create(objs, **kwargs)

    if not objs:
        return super().bulk_create(objs, **kwargs)

    clock = self.model.get_clock()

    for instance in objs:
        if not hasattr(instance, 'get_syncable_field_names'):
            continue

        now = clock.now()
        timestamps = dict(instance.sync_field_timestamps)

        for name in instance.get_syncable_field_names():
            if name not in timestamps:
                timestamps[name] = now

        instance.sync_field_timestamps = timestamps

        if not instance.sync_field_last_modified:
            instance.sync_field_last_modified = now

    return super().bulk_create(objs, **kwargs)

bulk_update

Source code in django_spire/contrib/sync/django/queryset.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def bulk_update(
    self,
    objs: list[Any],
    fields: list[str] | tuple[str, ...],
    **kwargs: Any,
) -> int:
    if _is_bypassed():
        return super().bulk_update(objs, fields, **kwargs)

    if not objs:
        return super().bulk_update(objs, fields, **kwargs)

    clock = self.model.get_clock()
    needs_extra = False

    for instance in objs:
        if not hasattr(instance, '_sync_exclude_fields'):
            continue

        syncable = set(instance.get_syncable_field_names())
        dirty = syncable & set(fields)

        if not dirty:
            continue

        now = clock.now()
        timestamps = dict(instance.sync_field_timestamps)

        for name in dirty:
            timestamps[name] = now

        instance.sync_field_timestamps = timestamps
        instance.sync_field_last_modified = now
        needs_extra = True

    if needs_extra:
        sync_fields = {
            'sync_field_timestamps',
            'sync_field_last_modified',
        }

        fields = list(set(fields) | sync_fields)

    return super().bulk_update(objs, fields, **kwargs)

sync_bypass

Source code in django_spire/contrib/sync/django/queryset.py 
18
19
20
21
22
23
24
25
26
@contextmanager
def sync_bypass() -> Iterator[None]:
    previous = getattr(_bypass, 'active', False)
    _bypass.active = True

    try:
        yield
    finally:
        _bypass.active = previous