Skip to content

many_to_many

django_spire.contrib.sync.django.storage.many_to_many

logger = logging.getLogger(__name__) module-attribute

ManyToManyApplier

Source code in django_spire/contrib/sync/django/storage/many_to_many.py 
19
20
def __init__(self, identity_field: str = 'id') -> None:
    self._identity_field = identity_field

apply

Source code in django_spire/contrib/sync/django/storage/many_to_many.py 
22
23
24
25
26
27
28
29
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
60
61
62
63
64
65
66
67
68
69
70
def apply(
    self,
    model: type[SyncableMixin],
    pending: dict[str, dict[str, list[Any]]],
) -> set[str]:
    skipped: set[str] = set()

    if not pending:
        return skipped

    identity_lookup = {f'{self._identity_field}__in': list(pending.keys())}

    instances_by_key = {
        str(getattr(instance, self._identity_field)): instance
        for instance in model.objects.filter(**identity_lookup)
    }

    with sync_bypass():
        for key in sorted(pending.keys()):
            many_to_many_data = pending[key]
            instance = instances_by_key.get(key)

            if instance is None:
                logger.warning(
                    'Skipping M2M relations for %s key=%s: '
                    'instance not found after upsert',
                    model._meta.label,
                    key,
                )

                skipped.add(key)

                continue

            for field_name, values in sorted(many_to_many_data.items()):
                try:
                    getattr(instance, field_name).set(values)
                except IntegrityError:
                    logger.exception(
                        'M2M set failed for %s:%s field=%s values=%s',
                        model._meta.label,
                        key,
                        field_name,
                        values,
                    )

                    raise

    return skipped