mirror of
https://github.com/IRBorisov/ConceptPortal.git
synced 2025-06-26 13:00:39 +03:00
387 lines
15 KiB
Python
387 lines
15 KiB
Python
''' Serializers for persistent data manipulation. '''
|
|
from typing import Optional, cast
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.db import transaction
|
|
from django.db.models import Q
|
|
from rest_framework import serializers
|
|
from rest_framework.serializers import PrimaryKeyRelatedField as PKField
|
|
|
|
from apps.library.models import LibraryItem
|
|
from apps.library.serializers import (
|
|
LibraryItemBaseSerializer,
|
|
LibraryItemDetailsSerializer,
|
|
LibraryItemReferenceSerializer
|
|
)
|
|
from apps.oss.models import Inheritance
|
|
from shared import messages as msg
|
|
|
|
from ..models import Constituenta, CstType, RSForm
|
|
from .basics import CstParseSerializer
|
|
from .io_pyconcept import PyConceptAdapter
|
|
|
|
|
|
class CstBaseSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Constituenta all data. '''
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = Constituenta
|
|
fields = '__all__'
|
|
read_only_fields = ('id',)
|
|
|
|
|
|
class CstSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Constituenta data. '''
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = Constituenta
|
|
fields = '__all__'
|
|
read_only_fields = ('id', 'schema', 'order', 'alias', 'cst_type', 'definition_resolved', 'term_resolved')
|
|
|
|
def update(self, instance: Constituenta, validated_data) -> Constituenta:
|
|
data = validated_data # Note: use alias for better code readability
|
|
definition: Optional[str] = data['definition_raw'] if 'definition_raw' in data else None
|
|
term: Optional[str] = data['term_raw'] if 'term_raw' in data else None
|
|
term_changed = 'term_forms' in data
|
|
schema = RSForm(instance.schema)
|
|
if definition is not None and definition != instance.definition_raw:
|
|
data['definition_resolved'] = schema.resolver().resolve(definition)
|
|
if term is not None and term != instance.term_raw:
|
|
data['term_resolved'] = schema.resolver().resolve(term)
|
|
if data['term_resolved'] != instance.term_resolved and 'term_forms' not in data:
|
|
data['term_forms'] = []
|
|
term_changed = data['term_resolved'] != instance.term_resolved
|
|
result: Constituenta = super().update(instance, data)
|
|
if term_changed:
|
|
schema.on_term_change([result.pk])
|
|
result.refresh_from_db()
|
|
schema.save()
|
|
return result
|
|
|
|
|
|
class CstDetailsSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Constituenta data including parse. '''
|
|
parse = CstParseSerializer()
|
|
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = Constituenta
|
|
fields = '__all__'
|
|
|
|
|
|
class CstCreateSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Constituenta creation. '''
|
|
insert_after = serializers.IntegerField(required=False, allow_null=True)
|
|
alias = serializers.CharField(max_length=8)
|
|
cst_type = serializers.ChoiceField(CstType.choices)
|
|
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = Constituenta
|
|
fields = \
|
|
'alias', 'cst_type', 'convention', \
|
|
'term_raw', 'definition_raw', 'definition_formal', \
|
|
'insert_after', 'term_forms'
|
|
|
|
|
|
class RSFormSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Detailed data for RSForm. '''
|
|
subscribers = serializers.ListField(
|
|
child=serializers.IntegerField()
|
|
)
|
|
editors = serializers.ListField(
|
|
child=serializers.IntegerField()
|
|
)
|
|
items = serializers.ListField(
|
|
child=CstSerializer()
|
|
)
|
|
inheritance = serializers.ListField(
|
|
child=serializers.ListField(child=serializers.IntegerField())
|
|
)
|
|
oss = serializers.ListField(
|
|
child=LibraryItemReferenceSerializer()
|
|
)
|
|
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = LibraryItem
|
|
fields = '__all__'
|
|
|
|
def to_representation(self, instance: LibraryItem) -> dict:
|
|
result = LibraryItemDetailsSerializer(instance).data
|
|
result['items'] = []
|
|
for cst in RSForm(instance).constituents().order_by('order'):
|
|
result['items'].append(CstSerializer(cst).data)
|
|
result['inheritance'] = []
|
|
for link in Inheritance.objects.filter(Q(child__schema=instance) | Q(parent__schema=instance)):
|
|
result['inheritance'].append([link.child.pk, link.parent.pk])
|
|
result['oss'] = []
|
|
for oss in LibraryItem.objects.filter(items__result=instance).only('alias'):
|
|
result['oss'].append({
|
|
'id': oss.pk,
|
|
'alias': oss.alias
|
|
})
|
|
return result
|
|
|
|
def to_versioned_data(self) -> dict:
|
|
''' Create serializable version representation without redundant data. '''
|
|
result = self.to_representation(cast(LibraryItem, self.instance))
|
|
del result['versions']
|
|
del result['subscribers']
|
|
del result['editors']
|
|
del result['inheritance']
|
|
del result['oss']
|
|
|
|
del result['owner']
|
|
del result['visible']
|
|
del result['read_only']
|
|
del result['access_policy']
|
|
del result['location']
|
|
|
|
del result['time_create']
|
|
del result['time_update']
|
|
return result
|
|
|
|
def from_versioned_data(self, version: int, data: dict) -> dict:
|
|
''' Load data from version. '''
|
|
result = self.to_representation(cast(LibraryItem, self.instance))
|
|
result['version'] = version
|
|
return result | data
|
|
|
|
@transaction.atomic
|
|
def restore_from_version(self, data: dict):
|
|
''' Load data from version. '''
|
|
schema = RSForm(cast(LibraryItem, self.instance))
|
|
items: list[dict] = data['items']
|
|
ids: list[int] = [item['id'] for item in items]
|
|
processed: list[int] = []
|
|
|
|
for cst in schema.constituents():
|
|
if not cst.pk in ids:
|
|
cst.delete()
|
|
else:
|
|
cst_data = next(x for x in items if x['id'] == cst.pk)
|
|
new_cst = CstBaseSerializer(data=cst_data)
|
|
new_cst.is_valid(raise_exception=True)
|
|
new_cst.update(
|
|
instance=cst,
|
|
validated_data=new_cst.validated_data
|
|
)
|
|
processed.append(cst.pk)
|
|
|
|
for cst_data in items:
|
|
if cst_data['id'] not in processed:
|
|
cst = schema.insert_new(cst_data['alias'])
|
|
cst_data['id'] = cst.pk
|
|
new_cst = CstBaseSerializer(data=cst_data)
|
|
new_cst.is_valid(raise_exception=True)
|
|
new_cst.update(
|
|
instance=cst,
|
|
validated_data=new_cst.validated_data
|
|
)
|
|
|
|
loaded_item = LibraryItemBaseSerializer(data=data)
|
|
loaded_item.is_valid(raise_exception=True)
|
|
loaded_item.update(
|
|
instance=cast(LibraryItem, self.instance),
|
|
validated_data=loaded_item.validated_data
|
|
)
|
|
|
|
|
|
class RSFormParseSerializer(serializers.ModelSerializer):
|
|
''' Serializer: Detailed data for RSForm including parse. '''
|
|
subscribers = serializers.ListField(
|
|
child=serializers.IntegerField()
|
|
)
|
|
editors = serializers.ListField(
|
|
child=serializers.IntegerField()
|
|
)
|
|
items = serializers.ListField(
|
|
child=CstDetailsSerializer()
|
|
)
|
|
|
|
class Meta:
|
|
''' serializer metadata. '''
|
|
model = LibraryItem
|
|
fields = '__all__'
|
|
|
|
def to_representation(self, instance: LibraryItem):
|
|
result = RSFormSerializer(instance).data
|
|
return self._parse_data(result)
|
|
|
|
def from_versioned_data(self, version: int, data: dict) -> dict:
|
|
''' Load data from version and parse. '''
|
|
item = cast(LibraryItem, self.instance)
|
|
result = RSFormSerializer(item).from_versioned_data(version, data)
|
|
return self._parse_data(result)
|
|
|
|
def _parse_data(self, data: dict) -> dict:
|
|
parse = PyConceptAdapter(data).parse()
|
|
for cst_data in data['items']:
|
|
cst_data['parse'] = next(
|
|
cst['parse'] for cst in parse['items']
|
|
if cst['id'] == cst_data['id']
|
|
)
|
|
return data
|
|
|
|
|
|
class CstTargetSerializer(serializers.Serializer):
|
|
''' Serializer: Target single Constituenta. '''
|
|
target = PKField(many=False, queryset=Constituenta.objects.all())
|
|
|
|
def validate(self, attrs):
|
|
schema = cast(LibraryItem, self.context['schema'])
|
|
cst = cast(Constituenta, attrs['target'])
|
|
if schema and cst.schema_id != schema.pk:
|
|
raise serializers.ValidationError({
|
|
f'{cst.pk}': msg.constituentaNotInRSform(schema.title)
|
|
})
|
|
if cst.cst_type not in [CstType.FUNCTION, CstType.STRUCTURED, CstType.TERM]:
|
|
raise serializers.ValidationError({
|
|
f'{cst.pk}': msg.constituentaNoStructure()
|
|
})
|
|
self.instance = cst
|
|
return attrs
|
|
|
|
|
|
class CstRenameSerializer(serializers.Serializer):
|
|
''' Serializer: Constituenta renaming. '''
|
|
target = PKField(many=False, queryset=Constituenta.objects.only('alias', 'schema'))
|
|
alias = serializers.CharField()
|
|
cst_type = serializers.CharField()
|
|
|
|
def validate(self, attrs):
|
|
attrs = super().validate(attrs)
|
|
schema = cast(LibraryItem, self.context['schema'])
|
|
cst = cast(Constituenta, attrs['target'])
|
|
if cst.schema_id != schema.pk:
|
|
raise serializers.ValidationError({
|
|
f'{cst.pk}': msg.constituentaNotInRSform(schema.title)
|
|
})
|
|
new_alias = self.initial_data['alias']
|
|
if cst.alias == new_alias:
|
|
raise serializers.ValidationError({
|
|
'alias': msg.renameTrivial(new_alias)
|
|
})
|
|
if RSForm(schema).constituents().filter(alias=new_alias).exists():
|
|
raise serializers.ValidationError({
|
|
'alias': msg.aliasTaken(new_alias)
|
|
})
|
|
return attrs
|
|
|
|
|
|
class CstListSerializer(serializers.Serializer):
|
|
''' Serializer: List of constituents from one origin. '''
|
|
items = PKField(many=True, queryset=Constituenta.objects.all())
|
|
|
|
def validate(self, attrs):
|
|
schema = cast(LibraryItem, self.context['schema'])
|
|
if not schema:
|
|
return attrs
|
|
|
|
for item in attrs['items']:
|
|
if item.schema_id != schema.pk:
|
|
raise serializers.ValidationError({
|
|
f'{item.pk}': msg.constituentaNotInRSform(schema.title)
|
|
})
|
|
return attrs
|
|
|
|
|
|
class CstMoveSerializer(CstListSerializer):
|
|
''' Serializer: Change constituenta position. '''
|
|
move_to = serializers.IntegerField()
|
|
|
|
|
|
class SubstitutionSerializerBase(serializers.Serializer):
|
|
''' Serializer: Basic substitution. '''
|
|
original = PKField(many=False, queryset=Constituenta.objects.only('alias', 'schema'))
|
|
substitution = PKField(many=False, queryset=Constituenta.objects.only('alias', 'schema'))
|
|
|
|
|
|
class CstSubstituteSerializer(serializers.Serializer):
|
|
''' Serializer: Constituenta substitution. '''
|
|
substitutions = serializers.ListField(
|
|
child=SubstitutionSerializerBase(),
|
|
min_length=1
|
|
)
|
|
|
|
def validate(self, attrs):
|
|
schema = cast(LibraryItem, self.context['schema'])
|
|
deleted = set()
|
|
for item in attrs['substitutions']:
|
|
original_cst = cast(Constituenta, item['original'])
|
|
substitution_cst = cast(Constituenta, item['substitution'])
|
|
if original_cst.pk in deleted:
|
|
raise serializers.ValidationError({
|
|
f'{original_cst.pk}': msg.substituteDouble(original_cst.alias)
|
|
})
|
|
if original_cst.alias == substitution_cst.alias:
|
|
raise serializers.ValidationError({
|
|
'alias': msg.substituteTrivial(original_cst.alias)
|
|
})
|
|
if original_cst.schema_id != schema.pk:
|
|
raise serializers.ValidationError({
|
|
'original': msg.constituentaNotInRSform(schema.title)
|
|
})
|
|
if substitution_cst.schema_id != schema.pk:
|
|
raise serializers.ValidationError({
|
|
'substitution': msg.constituentaNotInRSform(schema.title)
|
|
})
|
|
deleted.add(original_cst.pk)
|
|
return attrs
|
|
|
|
|
|
class InlineSynthesisSerializer(serializers.Serializer):
|
|
''' Serializer: Inline synthesis operation input. '''
|
|
receiver = PKField(many=False, queryset=LibraryItem.objects.all())
|
|
source = PKField(many=False, queryset=LibraryItem.objects.all()) # type: ignore
|
|
items = PKField(many=True, queryset=Constituenta.objects.all())
|
|
substitutions = serializers.ListField(
|
|
child=SubstitutionSerializerBase()
|
|
)
|
|
|
|
def validate(self, attrs):
|
|
user = cast(User, self.context['user'])
|
|
schema_in = cast(LibraryItem, attrs['source'])
|
|
schema_out = cast(LibraryItem, attrs['receiver'])
|
|
if user.is_anonymous or (schema_out.owner != user and not user.is_staff):
|
|
raise PermissionDenied({
|
|
'message': msg.schemaForbidden(),
|
|
'object_id': schema_in.pk
|
|
})
|
|
constituents = cast(list[Constituenta], attrs['items'])
|
|
for cst in constituents:
|
|
if cst.schema_id != schema_in.pk:
|
|
raise serializers.ValidationError({
|
|
f'{cst.pk}': msg.constituentaNotInRSform(schema_in.title)
|
|
})
|
|
deleted = set()
|
|
for item in attrs['substitutions']:
|
|
original_cst = cast(Constituenta, item['original'])
|
|
substitution_cst = cast(Constituenta, item['substitution'])
|
|
if original_cst.schema_id == schema_in.pk:
|
|
if original_cst not in constituents:
|
|
raise serializers.ValidationError({
|
|
f'{original_cst.pk}': msg.substitutionNotInList()
|
|
})
|
|
if substitution_cst.schema_id != schema_out.pk:
|
|
raise serializers.ValidationError({
|
|
f'{substitution_cst.pk}': msg.constituentaNotInRSform(schema_out.title)
|
|
})
|
|
else:
|
|
if substitution_cst not in constituents:
|
|
raise serializers.ValidationError({
|
|
f'{substitution_cst.pk}': msg.substitutionNotInList()
|
|
})
|
|
if original_cst.schema_id != schema_out.pk:
|
|
raise serializers.ValidationError({
|
|
f'{original_cst.pk}': msg.constituentaNotInRSform(schema_out.title)
|
|
})
|
|
if original_cst.pk in deleted:
|
|
raise serializers.ValidationError({
|
|
f'{original_cst.pk}': msg.substituteDouble(original_cst.alias)
|
|
})
|
|
deleted.add(original_cst.pk)
|
|
return attrs
|