Skip to content

serializer

django_spire.contrib.sync.database.serializer

__all__ = ['DjangoModelSerializer', 'Serializer'] module-attribute

Serializer

Bases: Protocol

deserialize

Source code in django_spire/contrib/sync/database/serializer/base.py 
10
def deserialize(self, model_label: str, record: SyncRecord) -> Any: ...

get_identity

Source code in django_spire/contrib/sync/database/serializer/base.py 
11
def get_identity(self, instance: Any) -> str: ...

serialize

Source code in django_spire/contrib/sync/database/serializer/base.py 
12
def serialize(self, instance: Any) -> SyncRecord: ...

DjangoModelSerializer

Bases: Serializer

Source code in django_spire/contrib/sync/database/serializer/django.py 
13
14
15
16
17
18
19
def __init__(
    self,
    models: list[type],
    identity_field: str = 'id',
) -> None:
    self._identity_field = identity_field
    self._models = {model._meta.label: model for model in models}

deserialize

Source code in django_spire/contrib/sync/database/serializer/django.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def deserialize(
    self,
    model_label: str,
    record: SyncRecord,
) -> Any:
    model = self._resolve_model(model_label)
    many_to_many_names = set(self._get_many_to_many_names(model))

    field_data = {
        key: value for key, value in record.data.items()
        if key not in many_to_many_names
    }

    instance = model(**field_data)
    instance.sync_field_timestamps = record.timestamps

    return instance

get_identity

Source code in django_spire/contrib/sync/database/serializer/django.py 
57
58
59
def get_identity(self, instance: Any) -> str:
    value = getattr(instance, self._identity_field)
    return str(value)

serialize

Source code in django_spire/contrib/sync/database/serializer/django.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def serialize(self, instance: Any) -> SyncRecord:
    data: dict[str, Any] = {}

    for field in instance._meta.concrete_fields:
        value = getattr(instance, field.attname)
        data[field.name] = self._coerce_value(value)

    for field in instance._meta.many_to_many:
        related = getattr(instance, field.name)

        data[field.name] = sorted(
            str(primary_key)
            for primary_key in related.values_list('pk', flat=True)
        )

    key = str(getattr(instance, self._identity_field))
    timestamps = dict(getattr(instance, 'sync_field_timestamps', {}))

    return SyncRecord(key=key, data=data, timestamps=timestamps)