69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
"""transfer common classes"""
|
|
from dataclasses import asdict, dataclass, field, is_dataclass
|
|
from json.encoder import JSONEncoder
|
|
from typing import Any, Dict, List
|
|
from uuid import UUID
|
|
|
|
from passbook.lib.models import SerializerModel
|
|
from passbook.lib.sentry import SentryIgnoredException
|
|
|
|
|
|
def get_attrs(obj: SerializerModel) -> Dict[str, Any]:
|
|
"""Get object's attributes via their serializer, and covert it to a normal dict"""
|
|
data = dict(obj.serializer(obj).data)
|
|
to_remove = ("policies", "stages", "pk")
|
|
for to_remove_name in to_remove:
|
|
if to_remove_name in data:
|
|
data.pop(to_remove_name)
|
|
return data
|
|
|
|
|
|
@dataclass
|
|
class FlowBundleEntry:
|
|
"""Single entry of a bundle"""
|
|
|
|
identifiers: Dict[str, Any]
|
|
model: str
|
|
attrs: Dict[str, Any]
|
|
|
|
@staticmethod
|
|
def from_model(
|
|
model: SerializerModel, *extra_identifier_names: str
|
|
) -> "FlowBundleEntry":
|
|
"""Convert a SerializerModel instance to a Bundle Entry"""
|
|
identifiers = {
|
|
"pk": model.pk,
|
|
}
|
|
all_attrs = get_attrs(model)
|
|
|
|
for extra_identifier_name in extra_identifier_names:
|
|
identifiers[extra_identifier_name] = all_attrs.pop(extra_identifier_name)
|
|
return FlowBundleEntry(
|
|
identifiers=identifiers,
|
|
model=f"{model._meta.app_label}.{model._meta.model_name}",
|
|
attrs=all_attrs,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FlowBundle:
|
|
"""Dataclass used for a full export"""
|
|
|
|
version: int = field(default=1)
|
|
entries: List[FlowBundleEntry] = field(default_factory=list)
|
|
|
|
|
|
class DataclassEncoder(JSONEncoder):
|
|
"""Convert FlowBundleEntry to json"""
|
|
|
|
def default(self, o):
|
|
if is_dataclass(o):
|
|
return asdict(o)
|
|
if isinstance(o, UUID):
|
|
return str(o)
|
|
return super().default(o)
|
|
|
|
|
|
class EntryInvalidError(SentryIgnoredException):
|
|
"""Error raised when an entry is invalid"""
|