Refactor: split backend rsform API into managable parts

This commit is contained in:
IRBorisov 2024-03-11 18:08:28 +03:00
parent e2a20ab91d
commit 1d701530df
33 changed files with 2732 additions and 2500 deletions

View File

@ -423,6 +423,7 @@ disable=too-many-public-methods,
no-else-continue,
no-else-return,
no-member,
too-many-ancestors,
too-many-return-statements,
too-many-locals,
too-many-instance-attributes,

View File

@ -0,0 +1,101 @@
''' Models: Constituenta. '''
from django.db.models import (
CASCADE, ForeignKey, Model, PositiveIntegerField,
TextChoices, TextField, CharField, JSONField
)
from django.core.validators import MinValueValidator
from django.urls import reverse
class CstType(TextChoices):
''' Type of constituenta '''
BASE = 'basic'
CONSTANT = 'constant'
STRUCTURED = 'structure'
AXIOM = 'axiom'
TERM = 'term'
FUNCTION = 'function'
PREDICATE = 'predicate'
THEOREM = 'theorem'
def _empty_forms():
return []
class Constituenta(Model):
''' Constituenta is the base unit for every conceptual schema '''
schema: ForeignKey = ForeignKey(
verbose_name='Концептуальная схема',
to='rsform.LibraryItem',
on_delete=CASCADE
)
order: PositiveIntegerField = PositiveIntegerField(
verbose_name='Позиция',
validators=[MinValueValidator(1)],
default=-1,
)
alias: CharField = CharField(
verbose_name='Имя',
max_length=8,
default='undefined'
)
cst_type: CharField = CharField(
verbose_name='Тип',
max_length=10,
choices=CstType.choices,
default=CstType.BASE
)
convention: TextField = TextField(
verbose_name='Комментарий/Конвенция',
default='',
blank=True
)
term_raw: TextField = TextField(
verbose_name='Термин (с отсылками)',
default='',
blank=True
)
term_resolved: TextField = TextField(
verbose_name='Термин',
default='',
blank=True
)
term_forms: JSONField = JSONField(
verbose_name='Словоформы',
default=_empty_forms
)
definition_formal: TextField = TextField(
verbose_name='Родоструктурное определение',
default='',
blank=True
)
definition_raw: TextField = TextField(
verbose_name='Текстовое определение (с отсылками)',
default='',
blank=True
)
definition_resolved: TextField = TextField(
verbose_name='Текстовое определение',
default='',
blank=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Конституента'
verbose_name_plural = 'Конституенты'
def get_absolute_url(self):
''' URL access. '''
return reverse('constituenta-detail', kwargs={'pk': self.pk})
def __str__(self) -> str:
return f'{self.alias}'
def set_term_resolved(self, new_term: str):
''' Set term and reset forms if needed. '''
if new_term == self.term_resolved:
return
self.term_resolved = new_term
self.term_forms = []

View File

@ -0,0 +1,85 @@
''' Models: LibraryItem. '''
from django.db import transaction
from django.db.models import (
SET_NULL, ForeignKey, Model,
TextChoices, TextField, BooleanField, CharField, DateTimeField
)
from apps.users.models import User
from .Version import Version
from .Subscription import Subscription
class LibraryItemType(TextChoices):
''' Type of library items '''
RSFORM = 'rsform'
OPERATIONS_SCHEMA = 'oss'
class LibraryItem(Model):
''' Abstract library item.'''
item_type: CharField = CharField(
verbose_name='Тип',
max_length=50,
choices=LibraryItemType.choices
)
owner: ForeignKey = ForeignKey(
verbose_name='Владелец',
to=User,
on_delete=SET_NULL,
null=True
)
title: TextField = TextField(
verbose_name='Название'
)
alias: CharField = CharField(
verbose_name='Шифр',
max_length=255,
blank=True
)
comment: TextField = TextField(
verbose_name='Комментарий',
blank=True
)
is_common: BooleanField = BooleanField(
verbose_name='Общая',
default=False
)
is_canonical: BooleanField = BooleanField(
verbose_name='Каноничная',
default=False
)
time_create: DateTimeField = DateTimeField(
verbose_name='Дата создания',
auto_now_add=True
)
time_update: DateTimeField = DateTimeField(
verbose_name='Дата изменения',
auto_now=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Схема'
verbose_name_plural = 'Схемы'
def __str__(self) -> str:
return f'{self.title}'
def get_absolute_url(self):
return f'/api/library/{self.pk}'
def subscribers(self) -> list[Subscription]:
''' Get all subscribers for this item. '''
return [subscription.user for subscription in Subscription.objects.filter(item=self.pk)]
def versions(self) -> list[Version]:
''' Get all Versions of this item. '''
return list(Version.objects.filter(item=self.pk).order_by('-time_create'))
@transaction.atomic
def save(self, *args, **kwargs):
subscribe = not self.pk and self.owner
super().save(*args, **kwargs)
if subscribe:
Subscription.subscribe(user=self.owner, item=self)

View File

@ -0,0 +1,19 @@
''' Models: LibraryTemplate. '''
from django.db.models import (
CASCADE, ForeignKey, Model
)
class LibraryTemplate(Model):
''' Template for library items and constituents. '''
lib_source: ForeignKey = ForeignKey(
verbose_name='Источник',
to='rsform.LibraryItem',
on_delete=CASCADE,
null=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Шаблон'
verbose_name_plural = 'Шаблоны'

View File

@ -0,0 +1,50 @@
''' Models: Subscription. '''
from typing import TYPE_CHECKING
from django.db.models import (
CASCADE, ForeignKey, Model
)
from apps.users.models import User
if TYPE_CHECKING:
from .LibraryItem import LibraryItem
class Subscription(Model):
''' User subscription to library item. '''
user: ForeignKey = ForeignKey(
verbose_name='Пользователь',
to=User,
on_delete=CASCADE
)
item: ForeignKey = ForeignKey(
verbose_name='Элемент',
to='rsform.LibraryItem',
on_delete=CASCADE
)
class Meta:
''' Model metadata. '''
verbose_name = 'Подписки'
verbose_name_plural = 'Подписка'
unique_together = [['user', 'item']]
def __str__(self) -> str:
return f'{self.user} -> {self.item}'
@staticmethod
def subscribe(user: User, item: 'LibraryItem') -> bool:
''' Add subscription. '''
if Subscription.objects.filter(user=user, item=item).exists():
return False
Subscription.objects.create(user=user, item=item)
return True
@staticmethod
def unsubscribe(user: User, item: 'LibraryItem') -> bool:
''' Remove subscription. '''
sub = Subscription.objects.filter(user=user, item=item)
if not sub.exists():
return False
sub.delete()
return True

View File

@ -0,0 +1,39 @@
''' Models: Version. '''
from django.db.models import (
CASCADE, ForeignKey, Model,
TextField, CharField, DateTimeField, JSONField
)
class Version(Model):
''' Library item version archive. '''
item: ForeignKey = ForeignKey(
verbose_name='Схема',
to='rsform.LibraryItem',
on_delete=CASCADE
)
version = CharField(
verbose_name='Версия',
max_length=20,
blank=False
)
description: TextField = TextField(
verbose_name='Описание',
blank=True
)
data: JSONField = JSONField(
verbose_name='Содержание'
)
time_create: DateTimeField = DateTimeField(
verbose_name='Дата создания',
auto_now_add=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Версии'
verbose_name_plural = 'Версия'
unique_together = [['item', 'version']]
def __str__(self) -> str:
return f'{self.item} v{self.version}'

View File

@ -0,0 +1,8 @@
''' Django: Models. '''
from .api_RSForm import RSForm
from .Constituenta import Constituenta, CstType, _empty_forms
from .LibraryItem import User, LibraryItem, LibraryItemType
from .LibraryTemplate import LibraryTemplate
from .Version import Version
from .Subscription import Subscription

View File

@ -1,55 +1,26 @@
''' Models: RSForms for conceptual schemas. '''
''' Models: RSForm API. '''
import re
from typing import Iterable, Optional, cast
from django.db import transaction
from django.db.models import (
CASCADE, SET_NULL, ForeignKey, Model, PositiveIntegerField, QuerySet,
TextChoices, TextField, BooleanField, CharField, DateTimeField, JSONField
)
from django.core.validators import MinValueValidator
from django.db.models import QuerySet
from django.core.exceptions import ValidationError
from django.urls import reverse
from apps.users.models import User
from cctext import Resolver, Entity, extract_entities, split_grams, TermForm
from .graph import Graph
from .utils import apply_pattern
from . import messages as msg
from .LibraryItem import LibraryItem, LibraryItemType
from .Constituenta import CstType, Constituenta
from .Version import Version
from ..graph import Graph
from ..utils import apply_pattern
from .. import messages as msg
_REF_ENTITY_PATTERN = re.compile(r'@{([^0-9\-].*?)\|.*?}')
_GLOBAL_ID_PATTERN = re.compile(r'([XCSADFPT][0-9]+)') # cspell:disable-line
class LibraryItemType(TextChoices):
''' Type of library items '''
RSFORM = 'rsform'
OPERATIONS_SCHEMA = 'oss'
class CstType(TextChoices):
''' Type of constituenta '''
BASE = 'basic'
CONSTANT = 'constant'
STRUCTURED = 'structure'
AXIOM = 'axiom'
TERM = 'term'
FUNCTION = 'function'
PREDICATE = 'predicate'
THEOREM = 'theorem'
class Syntax(TextChoices):
''' Syntax types '''
UNDEF = 'undefined'
ASCII = 'ascii'
MATH = 'math'
def _empty_forms():
return []
def _get_type_prefix(cst_type: CstType) -> str:
''' Get alias prefix. '''
if cst_type == CstType.BASE:
@ -71,244 +42,8 @@ def _get_type_prefix(cst_type: CstType) -> str:
return 'X'
class LibraryItem(Model):
''' Abstract library item.'''
item_type: CharField = CharField(
verbose_name='Тип',
max_length=50,
choices=LibraryItemType.choices
)
owner: ForeignKey = ForeignKey(
verbose_name='Владелец',
to=User,
on_delete=SET_NULL,
null=True
)
title: TextField = TextField(
verbose_name='Название'
)
alias: CharField = CharField(
verbose_name='Шифр',
max_length=255,
blank=True
)
comment: TextField = TextField(
verbose_name='Комментарий',
blank=True
)
is_common: BooleanField = BooleanField(
verbose_name='Общая',
default=False
)
is_canonical: BooleanField = BooleanField(
verbose_name='Каноничная',
default=False
)
time_create: DateTimeField = DateTimeField(
verbose_name='Дата создания',
auto_now_add=True
)
time_update: DateTimeField = DateTimeField(
verbose_name='Дата изменения',
auto_now=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Схема'
verbose_name_plural = 'Схемы'
def __str__(self) -> str:
return f'{self.title}'
def get_absolute_url(self):
return f'/api/library/{self.pk}'
def subscribers(self) -> list[User]:
''' Get all subscribers for this item. '''
return [subscription.user for subscription in Subscription.objects.filter(item=self.pk)]
def versions(self) -> list['Version']:
''' Get all Versions of this item. '''
return list(Version.objects.filter(item=self.pk).order_by('-time_create'))
@transaction.atomic
def save(self, *args, **kwargs):
subscribe = not self.pk and self.owner
super().save(*args, **kwargs)
if subscribe:
Subscription.subscribe(user=self.owner, item=self)
class LibraryTemplate(Model):
''' Template for library items and constituents. '''
lib_source: ForeignKey = ForeignKey(
verbose_name='Источник',
to=LibraryItem,
on_delete=CASCADE,
null=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Шаблон'
verbose_name_plural = 'Шаблоны'
class Version(Model):
''' Library item version archive. '''
item: ForeignKey = ForeignKey(
verbose_name='Схема',
to=LibraryItem,
on_delete=CASCADE
)
version = CharField(
verbose_name='Версия',
max_length=20,
blank=False
)
description: TextField = TextField(
verbose_name='Описание',
blank=True
)
data: JSONField = JSONField(
verbose_name='Содержание'
)
time_create: DateTimeField = DateTimeField(
verbose_name='Дата создания',
auto_now_add=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Версии'
verbose_name_plural = 'Версия'
unique_together = [['item', 'version']]
def __str__(self) -> str:
return f'{self.item} v{self.version}'
class Subscription(Model):
''' User subscription to library item. '''
user: ForeignKey = ForeignKey(
verbose_name='Пользователь',
to=User,
on_delete=CASCADE
)
item: ForeignKey = ForeignKey(
verbose_name='Элемент',
to=LibraryItem,
on_delete=CASCADE
)
class Meta:
''' Model metadata. '''
verbose_name = 'Подписки'
verbose_name_plural = 'Подписка'
unique_together = [['user', 'item']]
def __str__(self) -> str:
return f'{self.user} -> {self.item}'
@staticmethod
def subscribe(user: User, item: LibraryItem) -> bool:
''' Add subscription. '''
if Subscription.objects.filter(user=user, item=item).exists():
return False
Subscription.objects.create(user=user, item=item)
return True
@staticmethod
def unsubscribe(user: User, item: LibraryItem) -> bool:
''' Remove subscription. '''
sub = Subscription.objects.filter(user=user, item=item)
if not sub.exists():
return False
sub.delete()
return True
class Constituenta(Model):
''' Constituenta is the base unit for every conceptual schema '''
schema: ForeignKey = ForeignKey(
verbose_name='Концептуальная схема',
to=LibraryItem,
on_delete=CASCADE
)
order: PositiveIntegerField = PositiveIntegerField(
verbose_name='Позиция',
validators=[MinValueValidator(1)],
default=-1,
)
alias: CharField = CharField(
verbose_name='Имя',
max_length=8,
default='undefined'
)
cst_type: CharField = CharField(
verbose_name='Тип',
max_length=10,
choices=CstType.choices,
default=CstType.BASE
)
convention: TextField = TextField(
verbose_name='Комментарий/Конвенция',
default='',
blank=True
)
term_raw: TextField = TextField(
verbose_name='Термин (с отсылками)',
default='',
blank=True
)
term_resolved: TextField = TextField(
verbose_name='Термин',
default='',
blank=True
)
term_forms: JSONField = JSONField(
verbose_name='Словоформы',
default=_empty_forms
)
definition_formal: TextField = TextField(
verbose_name='Родоструктурное определение',
default='',
blank=True
)
definition_raw: TextField = TextField(
verbose_name='Текстовое определение (с отсылками)',
default='',
blank=True
)
definition_resolved: TextField = TextField(
verbose_name='Текстовое определение',
default='',
blank=True
)
class Meta:
''' Model metadata. '''
verbose_name = 'Конституента'
verbose_name_plural = 'Конституенты'
def get_absolute_url(self):
''' URL access. '''
return reverse('constituenta-detail', kwargs={'pk': self.pk})
def __str__(self) -> str:
return f'{self.alias}'
def set_term_resolved(self, new_term: str):
''' Set term and reset forms if needed. '''
if new_term == self.term_resolved:
return
self.term_resolved = new_term
self.term_forms = []
class RSForm:
''' RSForm is a math form of capturing conceptual schema. '''
''' RSForm is math form of conceptual schema. '''
def __init__(self, item: LibraryItem):
if item.item_type != LibraryItemType.RSFORM:
raise ValueError(msg.libraryTypeUnexpected())

View File

@ -1,730 +0,0 @@
''' Serializers for conceptual schema API. '''
import json
from typing import Optional, cast, Union
from rest_framework import serializers
from django.db import transaction
import pyconcept
from cctext import Resolver, Reference, ReferenceType, EntityReference, SyntacticReference
from .utils import fix_old_references
from .models import Constituenta, LibraryItem, RSForm, Version
from . import messages as msg
_CST_TYPE = 'constituenta'
_TRS_TYPE = 'rsform'
_TRS_VERSION_MIN = 16
_TRS_VERSION = 16
_TRS_HEADER = 'Exteor 4.8.13.1000 - 30/05/2022'
ConstituentaID = serializers.IntegerField
NodeID = serializers.IntegerField
class FileSerializer(serializers.Serializer):
''' Serializer: File input. '''
file = serializers.FileField(allow_empty_file=False)
class ExpressionSerializer(serializers.Serializer):
''' Serializer: RSLang expression. '''
expression = serializers.CharField()
class WordFormSerializer(serializers.Serializer):
''' Serializer: inflect request. '''
text = serializers.CharField()
grams = serializers.CharField()
class MultiFormSerializer(serializers.Serializer):
''' Serializer: inflect request. '''
items = serializers.ListField(
child=WordFormSerializer()
)
@staticmethod
def from_list(data: list[tuple[str, str]]) -> dict:
result: dict = {}
result['items'] = []
for item in data:
result['items'].append({
'text': item[0],
'grams': item[1]
})
return result
class TextSerializer(serializers.Serializer):
''' Serializer: Text with references. '''
text = serializers.CharField()
class LibraryItemSerializer(serializers.ModelSerializer):
''' Serializer: LibraryItem entry. '''
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
read_only_fields = ('owner', 'id', 'item_type')
class FunctionArgSerializer(serializers.Serializer):
''' Serializer: RSLang function argument type. '''
alias = serializers.CharField()
typification = serializers.CharField()
class CstParseSerializer(serializers.Serializer):
''' Serializer: Constituenta parse result. '''
status = serializers.CharField()
valueClass = serializers.CharField()
typification = serializers.CharField()
syntaxTree = serializers.CharField()
args = serializers.ListField(
child=FunctionArgSerializer()
)
class ErrorDescriptionSerializer(serializers.Serializer):
''' Serializer: RSError description. '''
errorType = serializers.IntegerField()
position = serializers.IntegerField()
isCritical = serializers.BooleanField()
params = serializers.ListField(
child=serializers.CharField()
)
class NodeDataSerializer(serializers.Serializer):
''' Serializer: Node data. '''
dataType = serializers.CharField()
value = serializers.CharField()
class ASTNodeSerializer(serializers.Serializer):
''' Serializer: Syntax tree node. '''
uid = NodeID()
parent = serializers.IntegerField() # type: ignore
typeID = serializers.IntegerField()
start = serializers.IntegerField()
finish = serializers.IntegerField()
data = NodeDataSerializer() # type: ignore
class ExpressionParseSerializer(serializers.Serializer):
''' Serializer: RSlang expression parse result. '''
parseResult = serializers.BooleanField()
syntax = serializers.CharField()
typification = serializers.CharField()
valueClass = serializers.CharField()
astText = serializers.CharField()
ast = serializers.ListField(
child=ASTNodeSerializer()
)
errors = serializers.ListField( # type: ignore
child=ErrorDescriptionSerializer()
)
args = serializers.ListField(
child=FunctionArgSerializer()
)
class VersionSerializer(serializers.ModelSerializer):
''' Serializer: Version data. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'item', 'description', 'time_create'
read_only_fields = ('id', 'item', 'time_create')
class VersionInnerSerializer(serializers.ModelSerializer):
''' Serializer: Version data for list of versions. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'description', 'time_create'
read_only_fields = ('id', 'item', 'time_create')
class VersionCreateSerializer(serializers.ModelSerializer):
''' Serializer: Version create data. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'version', 'description'
class LibraryItemDetailsSerializer(serializers.ModelSerializer):
''' Serializer: LibraryItem detailed data. '''
subscribers = serializers.SerializerMethodField()
versions = serializers.SerializerMethodField()
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
read_only_fields = ('owner', 'id', 'item_type')
def get_subscribers(self, instance: LibraryItem) -> list[int]:
return [item.pk for item in instance.subscribers()]
def get_versions(self, instance: LibraryItem) -> list:
return [VersionInnerSerializer(item).data for item in instance.versions()]
class ConstituentaSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta data. '''
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = '__all__'
read_only_fields = ('id', '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
schema = RSForm(instance.schema)
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
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.alias])
result.refresh_from_db()
schema.item.save()
return result
class CstCreateSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta creation. '''
insert_after = serializers.IntegerField(required=False, allow_null=True)
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = \
'alias', 'cst_type', 'convention', \
'term_raw', 'definition_raw', 'definition_formal', \
'insert_after', 'term_forms'
class CstRenameSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta renaming. '''
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = 'id', 'alias', 'cst_type'
def validate(self, attrs):
schema = cast(RSForm, self.context['schema'])
old_cst = Constituenta.objects.get(pk=self.initial_data['id'])
new_alias = self.initial_data['alias']
if old_cst.schema != schema.item:
raise serializers.ValidationError({
'id': msg.constituentaNotOwned(schema.item.title)
})
if old_cst.alias == new_alias:
raise serializers.ValidationError({
'alias': msg.renameTrivial(new_alias)
})
if schema.constituents().filter(alias=new_alias).exists():
raise serializers.ValidationError({
'alias': msg.renameTaken(new_alias)
})
self.instance = old_cst
attrs['schema'] = schema.item
attrs['id'] = self.initial_data['id']
return attrs
class CstSubstituteSerializer(serializers.Serializer):
''' Serializer: Constituenta substitution. '''
original = ConstituentaID()
substitution = ConstituentaID()
transfer_term = serializers.BooleanField(required=False, default=False)
def validate(self, attrs):
schema = cast(RSForm, self.context['schema'])
original_cst = Constituenta.objects.get(pk=self.initial_data['original'])
substitution_cst = Constituenta.objects.get(pk=self.initial_data['substitution'])
if original_cst.alias == substitution_cst.alias:
raise serializers.ValidationError({
'alias': msg.substituteTrivial(original_cst.alias)
})
if original_cst.schema != schema.item:
raise serializers.ValidationError({
'original': msg.constituentaNotOwned(schema.item.title)
})
if substitution_cst.schema != schema.item:
raise serializers.ValidationError({
'substitution': msg.constituentaNotOwned(schema.item.title)
})
attrs['original'] = original_cst
attrs['substitution'] = substitution_cst
attrs['transfer_term'] = self.initial_data['transfer_term']
return attrs
class CstListSerializer(serializers.Serializer):
''' Serializer: List of constituents from one origin. '''
items = serializers.ListField(
child=serializers.IntegerField()
)
def validate(self, attrs):
schema = self.context['schema']
cstList = []
for item in attrs['items']:
try:
cst = Constituenta.objects.get(pk=item)
except Constituenta.DoesNotExist as exception:
raise serializers.ValidationError({
f'{item}': msg.constituentaNotExists
}) from exception
if cst.schema != schema.item:
raise serializers.ValidationError({
f'{item}': msg.constituentaNotOwned(schema.item.title)
})
cstList.append(cst)
attrs['constituents'] = cstList
return attrs
class CstMoveSerializer(CstListSerializer):
''' Serializer: Change constituenta position. '''
move_to = serializers.IntegerField()
class TextPositionSerializer(serializers.Serializer):
''' Serializer: Text position. '''
start = serializers.IntegerField()
finish = serializers.IntegerField()
class ReferenceDataSerializer(serializers.Serializer):
''' Serializer: Reference data - Union of all references. '''
offset = serializers.IntegerField()
nominal = serializers.CharField()
entity = serializers.CharField()
form = serializers.CharField()
class ReferenceSerializer(serializers.Serializer):
''' Serializer: Language reference. '''
type = serializers.CharField()
data = ReferenceDataSerializer() # type: ignore
pos_input = TextPositionSerializer()
pos_output = TextPositionSerializer()
class ResolverSerializer(serializers.Serializer):
''' Serializer: Resolver results serializer. '''
input = serializers.CharField()
output = serializers.CharField()
refs = serializers.ListField(
child=ReferenceSerializer()
)
def to_representation(self, instance: Resolver) -> dict:
return {
'input': instance.input,
'output': instance.output,
'refs': [{
'type': ref.ref.get_type().value,
'data': self._get_reference_data(ref.ref),
'resolved': ref.resolved,
'pos_input': {
'start': ref.pos_input.start,
'finish': ref.pos_input.finish
},
'pos_output': {
'start': ref.pos_output.start,
'finish': ref.pos_output.finish
}
} for ref in instance.refs]
}
@staticmethod
def _get_reference_data(ref: Reference) -> dict:
if ref.get_type() == ReferenceType.entity:
return {
'entity': cast(EntityReference, ref).entity,
'form': cast(EntityReference, ref).form
}
else:
return {
'offset': cast(SyntacticReference, ref).offset,
'nominal': cast(SyntacticReference, ref).nominal
}
class PyConceptAdapter:
''' RSForm adapter for interacting with pyconcept module. '''
def __init__(self, data: Union[RSForm, dict]):
try:
if 'items' in cast(dict, data):
self.data = self._prepare_request_raw(cast(dict, data))
else:
self.data = self._prepare_request(cast(RSForm, data))
except TypeError:
self.data = self._prepare_request(cast(RSForm, data))
self._checked_data: Optional[dict] = None
def parse(self) -> dict:
''' Check RSForm and return check results.
Warning! Does not include texts. '''
self._produce_response()
if self._checked_data is None:
raise ValueError(msg.pyconceptFailure())
return self._checked_data
def _prepare_request(self, schema: RSForm) -> dict:
result: dict = {
'items': []
}
items = schema.constituents().order_by('order')
for cst in items:
result['items'].append({
'entityUID': cst.pk,
'cstType': cst.cst_type,
'alias': cst.alias,
'definition': {
'formal': cst.definition_formal
}
})
return result
def _prepare_request_raw(self, data: dict) -> dict:
result: dict = {
'items': []
}
for cst in data['items']:
result['items'].append({
'entityUID': cst['id'],
'cstType': cst['cst_type'],
'alias': cst['alias'],
'definition': {
'formal': cst['definition_formal']
}
})
return result
def _produce_response(self):
if self._checked_data is not None:
return
response = pyconcept.check_schema(json.dumps(self.data))
data = json.loads(response)
self._checked_data = {
'items': []
}
for cst in data['items']:
self._checked_data['items'].append({
'id': cst['entityUID'],
'cstType': cst['cstType'],
'alias': cst['alias'],
'definition': {
'formal': cst['definition']['formal']
},
'parse': cst['parse']
})
class RSFormSerializer(serializers.ModelSerializer):
''' Serializer: Detailed data for RSForm. '''
subscribers = serializers.ListField(
child=serializers.IntegerField()
)
items = serializers.ListField(
child=ConstituentaSerializer()
)
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
def to_representation(self, instance: LibraryItem) -> dict:
result = LibraryItemDetailsSerializer(instance).data
schema = RSForm(instance)
result['items'] = []
for cst in schema.constituents().order_by('order'):
result['items'].append(ConstituentaSerializer(cst).data)
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['owner']
del result['is_common']
del result['is_canonical']
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
class CstDetailsSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta data including parse. '''
parse = CstParseSerializer()
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = '__all__'
class RSFormParseSerializer(serializers.ModelSerializer):
''' Serializer: Detailed data for RSForm including parse. '''
subscribers = 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 RSFormUploadSerializer(serializers.Serializer):
''' Upload data for RSForm serializer. '''
file = serializers.FileField()
load_metadata = serializers.BooleanField()
class RSFormTRSSerializer(serializers.Serializer):
''' Serializer: TRS file production and loading for RSForm. '''
def to_representation(self, instance: RSForm) -> dict:
result = self._prepare_json_rsform(instance)
items = instance.constituents().order_by('order')
for cst in items:
result['items'].append(self._prepare_json_constituenta(cst))
return result
@staticmethod
def _prepare_json_rsform(schema: RSForm) -> dict:
return {
'type': _TRS_TYPE,
'title': schema.item.title,
'alias': schema.item.alias,
'comment': schema.item.comment,
'items': [],
'claimed': False,
'selection': [],
'version': _TRS_VERSION,
'versionInfo': _TRS_HEADER
}
@staticmethod
def _prepare_json_constituenta(cst: Constituenta) -> dict:
return {
'entityUID': cst.pk,
'type': _CST_TYPE,
'cstType': cst.cst_type,
'alias': cst.alias,
'convention': cst.convention,
'term': {
'raw': cst.term_raw,
'resolved': cst.term_resolved,
'forms': cst.term_forms
},
'definition': {
'formal': cst.definition_formal,
'text': {
'raw': cst.definition_raw,
'resolved': cst.definition_resolved
},
},
}
def from_versioned_data(self, data: dict) -> dict:
''' Load data from version. '''
result = {
'type': _TRS_TYPE,
'title': data['title'],
'alias': data['alias'],
'comment': data['comment'],
'items': [],
'claimed': False,
'selection': [],
'version': _TRS_VERSION,
'versionInfo': _TRS_HEADER
}
for cst in data['items']:
result['items'].append({
'entityUID': cst['id'],
'type': _CST_TYPE,
'cstType': cst['cst_type'],
'alias': cst['alias'],
'convention': cst['convention'],
'term': {
'raw': cst['term_raw'],
'resolved': cst['term_resolved'],
'forms': cst['term_forms']
},
'definition': {
'formal': cst['definition_formal'],
'text': {
'raw': cst['definition_raw'],
'resolved': cst['definition_resolved']
},
},
})
return result
def to_internal_value(self, data):
result = super().to_internal_value(data)
if 'owner' in data:
result['owner'] = data['owner']
if 'is_common' in data:
result['is_common'] = data['is_common']
if 'is_canonical' in data:
result['is_canonical'] = data['is_canonical']
result['items'] = data.get('items', [])
if self.context['load_meta']:
result['title'] = data.get('title', 'Без названия')
result['alias'] = data.get('alias', '')
result['comment']= data.get('comment', '')
if 'id' in data:
result['id'] = data['id']
self.instance = RSForm(LibraryItem.objects.get(pk=result['id']))
return result
def validate(self, attrs: dict):
if 'version' not in self.initial_data \
or self.initial_data['version'] < _TRS_VERSION_MIN \
or self.initial_data['version'] > _TRS_VERSION:
raise serializers.ValidationError({
'version': msg.exteorFileVersionNotSupported()
})
return attrs
@transaction.atomic
def create(self, validated_data: dict) -> RSForm:
self.instance: RSForm = RSForm.create(
owner=validated_data.get('owner', None),
alias=validated_data['alias'],
title=validated_data['title'],
comment=validated_data['comment'],
is_common=validated_data['is_common'],
is_canonical=validated_data['is_canonical']
)
self.instance.item.save()
order = 1
for cst_data in validated_data['items']:
cst = Constituenta(
alias=cst_data['alias'],
schema=self.instance.item,
order=order,
cst_type=cst_data['cstType'],
)
self._load_cst_texts(cst, cst_data)
cst.save()
order += 1
self.instance.resolve_all_text()
return self.instance
@transaction.atomic
def update(self, instance: RSForm, validated_data) -> RSForm:
if 'alias' in validated_data:
instance.item.alias = validated_data['alias']
if 'title' in validated_data:
instance.item.title = validated_data['title']
if 'comment' in validated_data:
instance.item.comment = validated_data['comment']
order = 1
prev_constituents = instance.constituents()
loaded_ids = set()
for cst_data in validated_data['items']:
uid = int(cst_data['entityUID'])
if prev_constituents.filter(pk=uid).exists():
cst: Constituenta = prev_constituents.get(pk=uid)
cst.order = order
cst.alias = cst_data['alias']
cst.cst_type = cst_data['cstType']
self._load_cst_texts(cst, cst_data)
cst.save()
else:
cst = Constituenta(
alias=cst_data['alias'],
schema=instance.item,
order=order,
cst_type=cst_data['cstType'],
)
self._load_cst_texts(cst, cst_data)
cst.save()
uid = cst.pk
loaded_ids.add(uid)
order += 1
for prev_cst in prev_constituents:
if prev_cst.pk not in loaded_ids:
prev_cst.delete()
instance.resolve_all_text()
instance.item.save()
return instance
@staticmethod
def _load_cst_texts(cst: Constituenta, data: dict):
cst.convention = data.get('convention', '')
if 'definition' in data:
cst.definition_formal = data['definition'].get('formal', '')
if 'text' in data['definition']:
cst.definition_raw = fix_old_references(data['definition']['text'].get('raw', ''))
else:
cst.definition_raw = ''
if 'term' in data:
cst.term_raw = fix_old_references(data['term'].get('raw', ''))
cst.term_forms = data['term'].get('forms', [])
else:
cst.term_raw = ''
cst.term_forms = []
class ResultTextResponse(serializers.Serializer):
''' Serializer: Text result of a function call. '''
result = serializers.CharField()
class NewCstResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
new_cst = ConstituentaSerializer()
schema = RSFormParseSerializer()
class NewVersionResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
version = serializers.IntegerField()
schema = RSFormParseSerializer()

View File

@ -0,0 +1,27 @@
''' REST API: Serializers. '''
from .basics import (
TextSerializer,
ExpressionSerializer,
ExpressionParseSerializer,
ResolverSerializer,
ASTNodeSerializer,
WordFormSerializer,
MultiFormSerializer
)
from .data_access import (
LibraryItemSerializer,
RSFormSerializer,
RSFormParseSerializer,
VersionSerializer,
VersionCreateSerializer,
ConstituentaSerializer,
CstMoveSerializer,
CstSubstituteSerializer,
CstCreateSerializer,
CstRenameSerializer,
CstListSerializer
)
from .schema_typing import (NewCstResponse, NewVersionResponse, ResultTextResponse)
from .io_pyconcept import PyConceptAdapter
from .io_files import (FileSerializer, RSFormUploadSerializer, RSFormTRSSerializer)

View File

@ -0,0 +1,166 @@
''' Basic serializers that do not interact with database. '''
from typing import cast
from rest_framework import serializers
from cctext import Resolver, Reference, ReferenceType, EntityReference, SyntacticReference
ConstituentaID = serializers.IntegerField
NodeID = serializers.IntegerField
class ExpressionSerializer(serializers.Serializer):
''' Serializer: RSLang expression. '''
expression = serializers.CharField()
class WordFormSerializer(serializers.Serializer):
''' Serializer: inflect request. '''
text = serializers.CharField()
grams = serializers.CharField()
class MultiFormSerializer(serializers.Serializer):
''' Serializer: inflect request. '''
items = serializers.ListField(
child=WordFormSerializer()
)
@staticmethod
def from_list(data: list[tuple[str, str]]) -> dict:
result: dict = {}
result['items'] = []
for item in data:
result['items'].append({
'text': item[0],
'grams': item[1]
})
return result
class TextSerializer(serializers.Serializer):
''' Serializer: Text with references. '''
text = serializers.CharField()
class FunctionArgSerializer(serializers.Serializer):
''' Serializer: RSLang function argument type. '''
alias = serializers.CharField()
typification = serializers.CharField()
class CstParseSerializer(serializers.Serializer):
''' Serializer: Constituenta parse result. '''
status = serializers.CharField()
valueClass = serializers.CharField()
typification = serializers.CharField()
syntaxTree = serializers.CharField()
args = serializers.ListField(
child=FunctionArgSerializer()
)
class ErrorDescriptionSerializer(serializers.Serializer):
''' Serializer: RSError description. '''
errorType = serializers.IntegerField()
position = serializers.IntegerField()
isCritical = serializers.BooleanField()
params = serializers.ListField(
child=serializers.CharField()
)
class NodeDataSerializer(serializers.Serializer):
''' Serializer: Node data. '''
dataType = serializers.CharField()
value = serializers.CharField()
class ASTNodeSerializer(serializers.Serializer):
''' Serializer: Syntax tree node. '''
uid = NodeID()
parent = serializers.IntegerField() # type: ignore
typeID = serializers.IntegerField()
start = serializers.IntegerField()
finish = serializers.IntegerField()
data = NodeDataSerializer() # type: ignore
class ExpressionParseSerializer(serializers.Serializer):
''' Serializer: RSlang expression parse result. '''
parseResult = serializers.BooleanField()
syntax = serializers.CharField()
typification = serializers.CharField()
valueClass = serializers.CharField()
astText = serializers.CharField()
ast = serializers.ListField(
child=ASTNodeSerializer()
)
errors = serializers.ListField( # type: ignore
child=ErrorDescriptionSerializer()
)
args = serializers.ListField(
child=FunctionArgSerializer()
)
class TextPositionSerializer(serializers.Serializer):
''' Serializer: Text position. '''
start = serializers.IntegerField()
finish = serializers.IntegerField()
class ReferenceDataSerializer(serializers.Serializer):
''' Serializer: Reference data - Union of all references. '''
offset = serializers.IntegerField()
nominal = serializers.CharField()
entity = serializers.CharField()
form = serializers.CharField()
class ReferenceSerializer(serializers.Serializer):
''' Serializer: Language reference. '''
type = serializers.CharField()
data = ReferenceDataSerializer() # type: ignore
pos_input = TextPositionSerializer()
pos_output = TextPositionSerializer()
class ResolverSerializer(serializers.Serializer):
''' Serializer: Resolver results serializer. '''
input = serializers.CharField()
output = serializers.CharField()
refs = serializers.ListField(
child=ReferenceSerializer()
)
def to_representation(self, instance: Resolver) -> dict:
return {
'input': instance.input,
'output': instance.output,
'refs': [{
'type': ref.ref.get_type().value,
'data': self._get_reference_data(ref.ref),
'resolved': ref.resolved,
'pos_input': {
'start': ref.pos_input.start,
'finish': ref.pos_input.finish
},
'pos_output': {
'start': ref.pos_output.start,
'finish': ref.pos_output.finish
}
} for ref in instance.refs]
}
@staticmethod
def _get_reference_data(ref: Reference) -> dict:
if ref.get_type() == ReferenceType.entity:
return {
'entity': cast(EntityReference, ref).entity,
'form': cast(EntityReference, ref).form
}
else:
return {
'offset': cast(SyntacticReference, ref).offset,
'nominal': cast(SyntacticReference, ref).nominal
}

View File

@ -0,0 +1,278 @@
''' Serializers for persistent data manipulation. '''
from typing import Optional, cast
from rest_framework import serializers
from .basics import ConstituentaID, CstParseSerializer
from .io_pyconcept import PyConceptAdapter
from ..models import Constituenta, LibraryItem, RSForm, Version
from .. import messages as msg
class LibraryItemSerializer(serializers.ModelSerializer):
''' Serializer: LibraryItem entry. '''
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
read_only_fields = ('owner', 'id', 'item_type')
class VersionSerializer(serializers.ModelSerializer):
''' Serializer: Version data. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'item', 'description', 'time_create'
read_only_fields = ('id', 'item', 'time_create')
class VersionInnerSerializer(serializers.ModelSerializer):
''' Serializer: Version data for list of versions. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'id', 'version', 'description', 'time_create'
read_only_fields = ('id', 'item', 'time_create')
class VersionCreateSerializer(serializers.ModelSerializer):
''' Serializer: Version create data. '''
class Meta:
''' serializer metadata. '''
model = Version
fields = 'version', 'description'
class LibraryItemDetailsSerializer(serializers.ModelSerializer):
''' Serializer: LibraryItem detailed data. '''
subscribers = serializers.SerializerMethodField()
versions = serializers.SerializerMethodField()
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
read_only_fields = ('owner', 'id', 'item_type')
def get_subscribers(self, instance: LibraryItem) -> list[int]:
return [item.pk for item in instance.subscribers()]
def get_versions(self, instance: LibraryItem) -> list:
return [VersionInnerSerializer(item).data for item in instance.versions()]
class ConstituentaSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta data. '''
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = '__all__'
read_only_fields = ('id', '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
schema = RSForm(instance.schema)
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
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.alias])
result.refresh_from_db()
schema.item.save()
return result
class CstCreateSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta creation. '''
insert_after = serializers.IntegerField(required=False, allow_null=True)
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = \
'alias', 'cst_type', 'convention', \
'term_raw', 'definition_raw', 'definition_formal', \
'insert_after', 'term_forms'
class CstRenameSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta renaming. '''
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = 'id', 'alias', 'cst_type'
def validate(self, attrs):
schema = cast(RSForm, self.context['schema'])
old_cst = Constituenta.objects.get(pk=self.initial_data['id'])
new_alias = self.initial_data['alias']
if old_cst.schema != schema.item:
raise serializers.ValidationError({
'id': msg.constituentaNotOwned(schema.item.title)
})
if old_cst.alias == new_alias:
raise serializers.ValidationError({
'alias': msg.renameTrivial(new_alias)
})
if schema.constituents().filter(alias=new_alias).exists():
raise serializers.ValidationError({
'alias': msg.renameTaken(new_alias)
})
self.instance = old_cst
attrs['schema'] = schema.item
attrs['id'] = self.initial_data['id']
return attrs
class RSFormSerializer(serializers.ModelSerializer):
''' Serializer: Detailed data for RSForm. '''
subscribers = serializers.ListField(
child=serializers.IntegerField()
)
items = serializers.ListField(
child=ConstituentaSerializer()
)
class Meta:
''' serializer metadata. '''
model = LibraryItem
fields = '__all__'
def to_representation(self, instance: LibraryItem) -> dict:
result = LibraryItemDetailsSerializer(instance).data
schema = RSForm(instance)
result['items'] = []
for cst in schema.constituents().order_by('order'):
result['items'].append(ConstituentaSerializer(cst).data)
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['owner']
del result['is_common']
del result['is_canonical']
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
class CstDetailsSerializer(serializers.ModelSerializer):
''' Serializer: Constituenta data including parse. '''
parse = CstParseSerializer()
class Meta:
''' serializer metadata. '''
model = Constituenta
fields = '__all__'
class RSFormParseSerializer(serializers.ModelSerializer):
''' Serializer: Detailed data for RSForm including parse. '''
subscribers = 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 CstSubstituteSerializer(serializers.Serializer):
''' Serializer: Constituenta substitution. '''
original = ConstituentaID()
substitution = ConstituentaID()
transfer_term = serializers.BooleanField(required=False, default=False)
def validate(self, attrs):
schema = cast(RSForm, self.context['schema'])
original_cst = Constituenta.objects.get(pk=self.initial_data['original'])
substitution_cst = Constituenta.objects.get(pk=self.initial_data['substitution'])
if original_cst.alias == substitution_cst.alias:
raise serializers.ValidationError({
'alias': msg.substituteTrivial(original_cst.alias)
})
if original_cst.schema != schema.item:
raise serializers.ValidationError({
'original': msg.constituentaNotOwned(schema.item.title)
})
if substitution_cst.schema != schema.item:
raise serializers.ValidationError({
'substitution': msg.constituentaNotOwned(schema.item.title)
})
attrs['original'] = original_cst
attrs['substitution'] = substitution_cst
attrs['transfer_term'] = self.initial_data['transfer_term']
return attrs
class CstListSerializer(serializers.Serializer):
''' Serializer: List of constituents from one origin. '''
items = serializers.ListField(
child=serializers.IntegerField()
)
def validate(self, attrs):
schema = self.context['schema']
cstList = []
for item in attrs['items']:
try:
cst = Constituenta.objects.get(pk=item)
except Constituenta.DoesNotExist as exception:
raise serializers.ValidationError({
f'{item}': msg.constituentaNotExists
}) from exception
if cst.schema != schema.item:
raise serializers.ValidationError({
f'{item}': msg.constituentaNotOwned(schema.item.title)
})
cstList.append(cst)
attrs['constituents'] = cstList
return attrs
class CstMoveSerializer(CstListSerializer):
''' Serializer: Change constituenta position. '''
move_to = serializers.IntegerField()

View File

@ -0,0 +1,213 @@
''' Serializers for file interaction. '''
from rest_framework import serializers
from django.db import transaction
from ..utils import fix_old_references
from ..models import Constituenta, LibraryItem, RSForm
from .. import messages as msg
_CST_TYPE = 'constituenta'
_TRS_TYPE = 'rsform'
_TRS_VERSION_MIN = 16
_TRS_VERSION = 16
_TRS_HEADER = 'Exteor 4.8.13.1000 - 30/05/2022'
class FileSerializer(serializers.Serializer):
''' Serializer: File input. '''
file = serializers.FileField(allow_empty_file=False)
class RSFormUploadSerializer(serializers.Serializer):
''' Upload data for RSForm serializer. '''
file = serializers.FileField()
load_metadata = serializers.BooleanField()
class RSFormTRSSerializer(serializers.Serializer):
''' Serializer: TRS file production and loading for RSForm. '''
def to_representation(self, instance: RSForm) -> dict:
result = self._prepare_json_rsform(instance)
items = instance.constituents().order_by('order')
for cst in items:
result['items'].append(self._prepare_json_constituenta(cst))
return result
@staticmethod
def _prepare_json_rsform(schema: RSForm) -> dict:
return {
'type': _TRS_TYPE,
'title': schema.item.title,
'alias': schema.item.alias,
'comment': schema.item.comment,
'items': [],
'claimed': False,
'selection': [],
'version': _TRS_VERSION,
'versionInfo': _TRS_HEADER
}
@staticmethod
def _prepare_json_constituenta(cst: Constituenta) -> dict:
return {
'entityUID': cst.pk,
'type': _CST_TYPE,
'cstType': cst.cst_type,
'alias': cst.alias,
'convention': cst.convention,
'term': {
'raw': cst.term_raw,
'resolved': cst.term_resolved,
'forms': cst.term_forms
},
'definition': {
'formal': cst.definition_formal,
'text': {
'raw': cst.definition_raw,
'resolved': cst.definition_resolved
},
},
}
def from_versioned_data(self, data: dict) -> dict:
''' Load data from version. '''
result = {
'type': _TRS_TYPE,
'title': data['title'],
'alias': data['alias'],
'comment': data['comment'],
'items': [],
'claimed': False,
'selection': [],
'version': _TRS_VERSION,
'versionInfo': _TRS_HEADER
}
for cst in data['items']:
result['items'].append({
'entityUID': cst['id'],
'type': _CST_TYPE,
'cstType': cst['cst_type'],
'alias': cst['alias'],
'convention': cst['convention'],
'term': {
'raw': cst['term_raw'],
'resolved': cst['term_resolved'],
'forms': cst['term_forms']
},
'definition': {
'formal': cst['definition_formal'],
'text': {
'raw': cst['definition_raw'],
'resolved': cst['definition_resolved']
},
},
})
return result
def to_internal_value(self, data):
result = super().to_internal_value(data)
if 'owner' in data:
result['owner'] = data['owner']
if 'is_common' in data:
result['is_common'] = data['is_common']
if 'is_canonical' in data:
result['is_canonical'] = data['is_canonical']
result['items'] = data.get('items', [])
if self.context['load_meta']:
result['title'] = data.get('title', 'Без названия')
result['alias'] = data.get('alias', '')
result['comment']= data.get('comment', '')
if 'id' in data:
result['id'] = data['id']
self.instance = RSForm(LibraryItem.objects.get(pk=result['id']))
return result
def validate(self, attrs: dict):
if 'version' not in self.initial_data \
or self.initial_data['version'] < _TRS_VERSION_MIN \
or self.initial_data['version'] > _TRS_VERSION:
raise serializers.ValidationError({
'version': msg.exteorFileVersionNotSupported()
})
return attrs
@transaction.atomic
def create(self, validated_data: dict) -> RSForm:
self.instance: RSForm = RSForm.create(
owner=validated_data.get('owner', None),
alias=validated_data['alias'],
title=validated_data['title'],
comment=validated_data['comment'],
is_common=validated_data['is_common'],
is_canonical=validated_data['is_canonical']
)
self.instance.item.save()
order = 1
for cst_data in validated_data['items']:
cst = Constituenta(
alias=cst_data['alias'],
schema=self.instance.item,
order=order,
cst_type=cst_data['cstType'],
)
self._load_cst_texts(cst, cst_data)
cst.save()
order += 1
self.instance.resolve_all_text()
return self.instance
@transaction.atomic
def update(self, instance: RSForm, validated_data) -> RSForm:
if 'alias' in validated_data:
instance.item.alias = validated_data['alias']
if 'title' in validated_data:
instance.item.title = validated_data['title']
if 'comment' in validated_data:
instance.item.comment = validated_data['comment']
order = 1
prev_constituents = instance.constituents()
loaded_ids = set()
for cst_data in validated_data['items']:
uid = int(cst_data['entityUID'])
if prev_constituents.filter(pk=uid).exists():
cst: Constituenta = prev_constituents.get(pk=uid)
cst.order = order
cst.alias = cst_data['alias']
cst.cst_type = cst_data['cstType']
self._load_cst_texts(cst, cst_data)
cst.save()
else:
cst = Constituenta(
alias=cst_data['alias'],
schema=instance.item,
order=order,
cst_type=cst_data['cstType'],
)
self._load_cst_texts(cst, cst_data)
cst.save()
uid = cst.pk
loaded_ids.add(uid)
order += 1
for prev_cst in prev_constituents:
if prev_cst.pk not in loaded_ids:
prev_cst.delete()
instance.resolve_all_text()
instance.item.save()
return instance
@staticmethod
def _load_cst_texts(cst: Constituenta, data: dict):
cst.convention = data.get('convention', '')
if 'definition' in data:
cst.definition_formal = data['definition'].get('formal', '')
if 'text' in data['definition']:
cst.definition_raw = fix_old_references(data['definition']['text'].get('raw', ''))
else:
cst.definition_raw = ''
if 'term' in data:
cst.term_raw = fix_old_references(data['term'].get('raw', ''))
cst.term_forms = data['term'].get('forms', [])
else:
cst.term_raw = ''
cst.term_forms = []

View File

@ -0,0 +1,79 @@
''' Data adapter to interface with pyconcept module. '''
import json
from typing import Optional, cast, Union
import pyconcept
from ..models import RSForm
from .. import messages as msg
class PyConceptAdapter:
''' RSForm adapter for interacting with pyconcept module. '''
def __init__(self, data: Union[RSForm, dict]):
try:
if 'items' in cast(dict, data):
self.data = self._prepare_request_raw(cast(dict, data))
else:
self.data = self._prepare_request(cast(RSForm, data))
except TypeError:
self.data = self._prepare_request(cast(RSForm, data))
self._checked_data: Optional[dict] = None
def parse(self) -> dict:
''' Check RSForm and return check results.
Warning! Does not include texts. '''
self._produce_response()
if self._checked_data is None:
raise ValueError(msg.pyconceptFailure())
return self._checked_data
def _prepare_request(self, schema: RSForm) -> dict:
result: dict = {
'items': []
}
items = schema.constituents().order_by('order')
for cst in items:
result['items'].append({
'entityUID': cst.pk,
'cstType': cst.cst_type,
'alias': cst.alias,
'definition': {
'formal': cst.definition_formal
}
})
return result
def _prepare_request_raw(self, data: dict) -> dict:
result: dict = {
'items': []
}
for cst in data['items']:
result['items'].append({
'entityUID': cst['id'],
'cstType': cst['cst_type'],
'alias': cst['alias'],
'definition': {
'formal': cst['definition_formal']
}
})
return result
def _produce_response(self):
if self._checked_data is not None:
return
response = pyconcept.check_schema(json.dumps(self.data))
data = json.loads(response)
self._checked_data = {
'items': []
}
for cst in data['items']:
self._checked_data['items'].append({
'id': cst['entityUID'],
'cstType': cst['cstType'],
'alias': cst['alias'],
'definition': {
'formal': cst['definition']['formal']
},
'parse': cst['parse']
})

View File

@ -0,0 +1,19 @@
''' Utility serializers for REST API schema - SHOULD NOT BE ACCESSED DIRECTLY. '''
from rest_framework import serializers
from .data_access import ConstituentaSerializer, RSFormParseSerializer
class ResultTextResponse(serializers.Serializer):
''' Serializer: Text result of a function call. '''
result = serializers.CharField()
class NewCstResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
new_cst = ConstituentaSerializer()
schema = RSFormParseSerializer()
class NewVersionResponse(serializers.Serializer):
''' Serializer: Create cst response. '''
version = serializers.IntegerField()
schema = RSFormParseSerializer()

View File

@ -1,6 +1,6 @@
''' Tests. '''
from .t_imports import *
from .t_views import *
from .s_views import *
from .t_models import *
from .t_serializers import *
from .t_graph import *

View File

@ -0,0 +1,8 @@
''' Tests for REST API. '''
from .t_library import *
from .t_constituents import *
from .t_rsforms import *
from .t_versions import *
from .t_cctext import *
from .t_rslang import *

View File

@ -0,0 +1,63 @@
''' Testing views '''
import os
import io
from zipfile import ZipFile
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework.exceptions import ErrorDetail
from rest_framework import status
from cctext import ReferenceType, split_grams
from apps.users.models import User
from apps.rsform.models import (
RSForm, Constituenta, CstType,
LibraryItem, LibraryItemType, Subscription, LibraryTemplate
)
from apps.rsform.views import (
convert_to_ascii,
convert_to_math,
parse_expression,
inflect,
parse_text,
generate_lexeme
)
class TestNaturalLanguageViews(APITestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
def _assert_tags(self, actual: str, expected: str):
self.assertEqual(set(split_grams(actual)), set(split_grams(expected)))
def test_parse_text(self):
data = {'text': 'синим слонам'}
request = self.factory.post(
'/api/cctext/parse',
data=data, format='json'
)
response = parse_text(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self._assert_tags(response.data['result'], 'datv,NOUN,plur,anim,masc')
def test_inflect(self):
data = {'text': 'синий слон', 'grams': 'plur,datv'}
request = self.factory.post(
'/api/cctext/inflect',
data=data, format='json'
)
response = inflect(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['result'], 'синим слонам')
def test_generate_lexeme(self):
data = {'text': 'синий слон'}
request = self.factory.post(
'/api/cctext/generate-lexeme',
data=data, format='json'
)
response = generate_lexeme(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['items']), 12)
self.assertEqual(response.data['items'][0]['text'], 'синий слон')

View File

@ -0,0 +1,120 @@
''' Testing API: Constituents. '''
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework import status
from apps.users.models import User
from apps.rsform.models import RSForm, Constituenta, CstType
class TestConstituentaAPI(APITestCase):
''' Testing Constituenta view. '''
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create(username='UserTest')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.rsform_owned = RSForm.create(title='Test', alias='T1', owner=self.user)
self.rsform_unowned = RSForm.create(title='Test2', alias='T2')
self.cst1 = Constituenta.objects.create(
alias='X1',
schema=self.rsform_owned.item,
order=1,
convention='Test',
term_raw='Test1',
term_resolved='Test1R',
term_forms=[{'text':'form1', 'tags':'sing,datv'}])
self.cst2 = Constituenta.objects.create(
alias='X2',
schema=self.rsform_unowned.item,
order=1,
convention='Test1',
term_raw='Test2',
term_resolved='Test2R'
)
self.cst3 = Constituenta.objects.create(
alias='X3',
schema=self.rsform_owned.item,
order=2,
term_raw='Test3',
term_resolved='Test3',
definition_raw='Test1',
definition_resolved='Test2'
)
def test_retrieve(self):
response = self.client.get(f'/api/constituents/{self.cst1.id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['alias'], self.cst1.alias)
self.assertEqual(response.data['convention'], self.cst1.convention)
def test_partial_update(self):
data = {'convention': 'tt'}
response = self.client.patch(
f'/api/constituents/{self.cst2.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.logout()
response = self.client.patch(
f'/api/constituents/{self.cst1.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(user=self.user)
response = self.client.patch(
f'/api/constituents/{self.cst1.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.cst1.refresh_from_db()
self.assertEqual(response.data['convention'], 'tt')
self.assertEqual(self.cst1.convention, 'tt')
response = self.client.patch(
f'/api/constituents/{self.cst1.id}',
data=data,
format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_update_resolved_no_refs(self):
data = {
'term_raw': 'New term',
'definition_raw': 'New def'
}
response = self.client.patch(f'/api/constituents/{self.cst3.id}', data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.cst3.refresh_from_db()
self.assertEqual(response.data['term_resolved'], 'New term')
self.assertEqual(self.cst3.term_resolved, 'New term')
self.assertEqual(response.data['definition_resolved'], 'New def')
self.assertEqual(self.cst3.definition_resolved, 'New def')
def test_update_resolved_refs(self):
data = {
'term_raw': '@{X1|nomn,sing}',
'definition_raw': '@{X1|nomn,sing} @{X1|sing,datv}'
}
response = self.client.patch(
f'/api/constituents/{self.cst3.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.cst3.refresh_from_db()
self.assertEqual(self.cst3.term_resolved, self.cst1.term_resolved)
self.assertEqual(response.data['term_resolved'], self.cst1.term_resolved)
self.assertEqual(self.cst3.definition_resolved, f'{self.cst1.term_resolved} form1')
self.assertEqual(response.data['definition_resolved'], f'{self.cst1.term_resolved} form1')
def test_readonly_cst_fields(self):
data = {'alias': 'X33', 'order': 10}
response = self.client.patch(
f'/api/constituents/{self.cst1.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['alias'], 'X1')
self.assertEqual(response.data['alias'], self.cst1.alias)
self.assertEqual(response.data['order'], self.cst1.order)

View File

@ -0,0 +1,164 @@
''' Testing API: Library. '''
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework import status
from apps.users.models import User
from apps.rsform.models import LibraryItem, LibraryItemType, Subscription, LibraryTemplate
from ..utils import response_contains
class TestLibraryViewset(APITestCase):
''' Testing Library view. '''
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create(username='UserTest')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.owned = LibraryItem.objects.create(
item_type=LibraryItemType.RSFORM,
title='Test',
alias='T1',
owner=self.user
)
self.unowned = LibraryItem.objects.create(
item_type=LibraryItemType.RSFORM,
title='Test2',
alias='T2'
)
self.common = LibraryItem.objects.create(
item_type=LibraryItemType.RSFORM,
title='Test3',
alias='T3',
is_common=True
)
def test_create_anonymous(self):
self.client.logout()
data = {'title': 'Title'}
response = self.client.post('/api/library', data=data, format='json')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_create_populate_user(self):
data = {'title': 'Title'}
response = self.client.post('/api/library', data=data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['title'], 'Title')
self.assertEqual(response.data['owner'], self.user.id)
def test_update(self):
data = {'id': self.owned.id, 'title': 'New title'}
response = self.client.patch(
f'/api/library/{self.owned.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['title'], 'New title')
self.assertEqual(response.data['alias'], self.owned.alias)
def test_update_unowned(self):
data = {'id': self.unowned.id, 'title': 'New title'}
response = self.client.patch(
f'/api/library/{self.unowned.id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_destroy(self):
response = self.client.delete(f'/api/library/{self.owned.id}')
self.assertTrue(response.status_code in [status.HTTP_202_ACCEPTED, status.HTTP_204_NO_CONTENT])
def test_destroy_admin_override(self):
response = self.client.delete(f'/api/library/{self.unowned.id}')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.user.is_staff = True
self.user.save()
response = self.client.delete(f'/api/library/{self.unowned.id}')
self.assertTrue(response.status_code in [status.HTTP_202_ACCEPTED, status.HTTP_204_NO_CONTENT])
def test_claim(self):
response = self.client.post(f'/api/library/{self.owned.id}/claim')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.owned.is_common = True
self.owned.save()
response = self.client.post(f'/api/library/{self.owned.id}/claim')
self.assertEqual(response.status_code, status.HTTP_304_NOT_MODIFIED)
response = self.client.post(f'/api/library/{self.unowned.id}/claim')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertFalse(self.user in self.unowned.subscribers())
self.unowned.is_common = True
self.unowned.save()
response = self.client.post(f'/api/library/{self.unowned.id}/claim')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.unowned.refresh_from_db()
self.assertEqual(self.unowned.owner, self.user)
self.assertEqual(self.unowned.owner, self.user)
self.assertTrue(self.user in self.unowned.subscribers())
def test_claim_anonymous(self):
self.client.logout()
response = self.client.post(f'/api/library/{self.owned.id}/claim')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_retrieve_common(self):
self.client.logout()
response = self.client.get('/api/library/active')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response_contains(response, self.common))
self.assertFalse(response_contains(response, self.unowned))
self.assertFalse(response_contains(response, self.owned))
def test_retrieve_owned(self):
response = self.client.get('/api/library/active')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response_contains(response, self.common))
self.assertFalse(response_contains(response, self.unowned))
self.assertTrue(response_contains(response, self.owned))
def test_retrieve_subscribed(self):
response = self.client.get('/api/library/active')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(response_contains(response, self.unowned))
user2 = User.objects.create(username='UserTest2')
Subscription.subscribe(user=self.user, item=self.unowned)
Subscription.subscribe(user=user2, item=self.unowned)
Subscription.subscribe(user=user2, item=self.owned)
response = self.client.get('/api/library/active')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response_contains(response, self.unowned))
self.assertEqual(len(response.data), 3)
def test_subscriptions(self):
response = self.client.delete(f'/api/library/{self.unowned.id}/unsubscribe')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(self.user in self.unowned.subscribers())
response = self.client.post(f'/api/library/{self.unowned.id}/subscribe')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertTrue(self.user in self.unowned.subscribers())
response = self.client.post(f'/api/library/{self.unowned.id}/subscribe')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertTrue(self.user in self.unowned.subscribers())
response = self.client.delete(f'/api/library/{self.unowned.id}/unsubscribe')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(self.user in self.unowned.subscribers())
def test_retrieve_templates(self):
response = self.client.get('/api/library/templates')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(response_contains(response, self.common))
self.assertFalse(response_contains(response, self.unowned))
self.assertFalse(response_contains(response, self.owned))
LibraryTemplate.objects.create(lib_source=self.unowned)
response = self.client.get('/api/library/templates')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(response_contains(response, self.common))
self.assertTrue(response_contains(response, self.unowned))
self.assertFalse(response_contains(response, self.owned))

View File

@ -0,0 +1,500 @@
''' Testing API: RSForms. '''
import os
import io
from zipfile import ZipFile
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework import status
from apps.users.models import User
from apps.rsform.models import (
RSForm,
Constituenta,
CstType,
LibraryItem,
LibraryItemType
)
from cctext import ReferenceType
from ..utils import response_contains
class TestRSFormViewset(APITestCase):
''' Testing RSForm view. '''
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create(username='UserTest')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.owned = RSForm.create(title='Test', alias='T1', owner=self.user)
self.unowned = RSForm.create(title='Test2', alias='T2')
def test_list(self):
non_schema = LibraryItem.objects.create(
item_type=LibraryItemType.OPERATIONS_SCHEMA,
title='Test3'
)
response = self.client.get('/api/rsforms')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(response_contains(response, non_schema))
self.assertTrue(response_contains(response, self.unowned.item))
self.assertTrue(response_contains(response, self.owned.item))
response = self.client.get('/api/library')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response_contains(response, non_schema))
self.assertTrue(response_contains(response, self.unowned.item))
self.assertTrue(response_contains(response, self.owned.item))
def test_contents(self):
schema = RSForm.create(title='Title1')
schema.insert_last(alias='X1', insert_type=CstType.BASE)
response = self.client.get(f'/api/rsforms/{schema.item.id}/contents')
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_details(self):
schema = RSForm.create(title='Test', owner=self.user)
x1 = schema.insert_at(1, 'X1', CstType.BASE)
x2 = schema.insert_at(2, 'X2', CstType.BASE)
x1.term_raw = 'человек'
x1.term_resolved = 'человек'
x2.term_raw = '@{X1|plur}'
x2.term_resolved = 'люди'
x1.save()
x2.save()
response = self.client.get(f'/api/rsforms/{schema.item.id}/details')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['title'], 'Test')
self.assertEqual(len(response.data['items']), 2)
self.assertEqual(response.data['items'][0]['id'], x1.id)
self.assertEqual(response.data['items'][0]['parse']['status'], 'verified')
self.assertEqual(response.data['items'][0]['term_raw'], x1.term_raw)
self.assertEqual(response.data['items'][0]['term_resolved'], x1.term_resolved)
self.assertEqual(response.data['items'][1]['id'], x2.id)
self.assertEqual(response.data['items'][1]['term_raw'], x2.term_raw)
self.assertEqual(response.data['items'][1]['term_resolved'], x2.term_resolved)
self.assertEqual(response.data['subscribers'], [self.user.pk])
def test_check(self):
schema = RSForm.create(title='Test')
schema.insert_at(1, 'X1', CstType.BASE)
data = {'expression': 'X1=X1'}
response = self.client.post(
f'/api/rsforms/{schema.item.id}/check',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['parseResult'], True)
self.assertEqual(response.data['syntax'], 'math')
self.assertEqual(response.data['astText'], '[=[X1][X1]]')
self.assertEqual(response.data['typification'], 'LOGIC')
self.assertEqual(response.data['valueClass'], 'value')
response = self.client.post(
f'/api/rsforms/{self.unowned.item.id}/check',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_resolve(self):
schema = RSForm.create(title='Test')
x1 = schema.insert_at(1, 'X1', CstType.BASE)
x1.term_resolved = 'синий слон'
x1.save()
data = {'text': '@{1|редкий} @{X1|plur,datv}'}
response = self.client.post(
f'/api/rsforms/{schema.item.id}/resolve',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['input'], '@{1|редкий} @{X1|plur,datv}')
self.assertEqual(response.data['output'], 'редким синим слонам')
self.assertEqual(len(response.data['refs']), 2)
self.assertEqual(response.data['refs'][0]['type'], ReferenceType.syntactic.value)
self.assertEqual(response.data['refs'][0]['resolved'], 'редким')
self.assertEqual(response.data['refs'][0]['data']['offset'], 1)
self.assertEqual(response.data['refs'][0]['data']['nominal'], 'редкий')
self.assertEqual(response.data['refs'][0]['pos_input']['start'], 0)
self.assertEqual(response.data['refs'][0]['pos_input']['finish'], 11)
self.assertEqual(response.data['refs'][0]['pos_output']['start'], 0)
self.assertEqual(response.data['refs'][0]['pos_output']['finish'], 6)
self.assertEqual(response.data['refs'][1]['type'], ReferenceType.entity.value)
self.assertEqual(response.data['refs'][1]['resolved'], 'синим слонам')
self.assertEqual(response.data['refs'][1]['data']['entity'], 'X1')
self.assertEqual(response.data['refs'][1]['data']['form'], 'plur,datv')
self.assertEqual(response.data['refs'][1]['pos_input']['start'], 12)
self.assertEqual(response.data['refs'][1]['pos_input']['finish'], 27)
self.assertEqual(response.data['refs'][1]['pos_output']['start'], 7)
self.assertEqual(response.data['refs'][1]['pos_output']['finish'], 19)
def test_import_trs(self):
work_dir = os.path.dirname(os.path.abspath(__file__))
with open(f'{work_dir}/data/sample-rsform.trs', 'rb') as file:
data = {'file': file}
response = self.client.post('/api/rsforms/import-trs', data=data, format='multipart')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['owner'], self.user.pk)
self.assertTrue(response.data['title'] != '')
def test_export_trs(self):
schema = RSForm.create(title='Test')
schema.insert_at(1, 'X1', CstType.BASE)
response = self.client.get(f'/api/rsforms/{schema.item.id}/export-trs')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.headers['Content-Disposition'], 'attachment; filename=Schema.trs')
with io.BytesIO(response.content) as stream:
with ZipFile(stream, 'r') as zipped_file:
self.assertIsNone(zipped_file.testzip())
self.assertIn('document.json', zipped_file.namelist())
def test_create_constituenta(self):
data = {'alias': 'X3', 'cst_type': 'basic'}
response = self.client.post(
f'/api/rsforms/{self.unowned.item.id}/cst-create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
item = self.owned.item
Constituenta.objects.create(
schema=item,
alias='X1',
cst_type='basic',
order=1
)
x2 = Constituenta.objects.create(
schema=item,
alias='X2',
cst_type='basic',
order=2
)
response = self.client.post(
f'/api/rsforms/{item.id}/cst-create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['new_cst']['alias'], 'X3')
x3 = Constituenta.objects.get(alias=response.data['new_cst']['alias'])
self.assertEqual(x3.order, 3)
data = {
'alias': 'X4',
'cst_type': 'basic',
'insert_after': x2.id,
'term_raw': 'test',
'term_forms': [{'text':'form1', 'tags':'sing,datv'}]
}
response = self.client.post(
f'/api/rsforms/{item.id}/cst-create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['new_cst']['alias'], data['alias'])
x4 = Constituenta.objects.get(alias=response.data['new_cst']['alias'])
self.assertEqual(x4.order, 3)
self.assertEqual(x4.term_raw, data['term_raw'])
self.assertEqual(x4.term_forms, data['term_forms'])
def test_rename_constituenta(self):
cst1 = Constituenta.objects.create(
alias='X1',
schema=self.owned.item,
order=1,
convention='Test',
term_raw='Test1',
term_resolved='Test1',
term_forms=[{'text':'form1', 'tags':'sing,datv'}]
)
cst2 = Constituenta.objects.create(
alias='X2',
schema=self.unowned.item,
order=1
)
cst3 = Constituenta.objects.create(
alias='X3',
schema=self.owned.item, order=2,
term_raw='Test3',
term_resolved='Test3',
definition_raw='Test1',
definition_resolved='Test2'
)
data = {'id': cst2.pk, 'alias': 'D2', 'cst_type': 'term'}
response = self.client.patch(
f'/api/rsforms/{self.unowned.item.id}/cst-rename',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-rename',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'id': cst1.pk, 'alias': cst1.alias, 'cst_type': 'term'}
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-rename',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'id': cst1.pk, 'alias': cst3.alias}
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-rename',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'alias': 'D2', 'cst_type': 'term', 'id': cst1.pk}
item = self.owned.item
d1 = Constituenta.objects.create(schema=item, alias='D1', cst_type='term', order=4)
d1.term_raw = '@{X1|plur}'
d1.definition_formal = 'X1'
d1.save()
self.assertEqual(d1.order, 4)
self.assertEqual(cst1.order, 1)
self.assertEqual(cst1.alias, 'X1')
self.assertEqual(cst1.cst_type, CstType.BASE)
response = self.client.patch(
f'/api/rsforms/{item.id}/cst-rename',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['new_cst']['alias'], 'D2')
self.assertEqual(response.data['new_cst']['cst_type'], 'term')
d1.refresh_from_db()
cst1.refresh_from_db()
self.assertEqual(d1.order, 4)
self.assertEqual(d1.term_resolved, '')
self.assertEqual(d1.term_raw, '@{D2|plur}')
self.assertEqual(cst1.order, 1)
self.assertEqual(cst1.alias, 'D2')
self.assertEqual(cst1.cst_type, CstType.TERM)
def test_substitute_constituenta(self):
x1 = Constituenta.objects.create(
alias='X1',
schema=self.owned.item,
order=1,
term_raw='Test1',
term_resolved='Test1',
term_forms=[{'text':'form1', 'tags':'sing,datv'}]
)
x2 = Constituenta.objects.create(
alias='X2',
schema=self.owned.item,
order=2,
term_raw='Test2'
)
unowned = Constituenta.objects.create(
alias='X2',
schema=self.unowned.item,
order=1
)
data = {'original': x1.pk, 'substitution': unowned.pk, 'transfer_term': True}
response = self.client.patch(
f'/api/rsforms/{self.unowned.item.id}/cst-substitute',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-substitute',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'original': unowned.pk, 'substitution': x1.pk, 'transfer_term': True}
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-substitute',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
data = {'original': x1.pk, 'substitution': x1.pk, 'transfer_term': True}
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-substitute',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
d1 = Constituenta.objects.create(
alias='D1',
schema=self.owned.item,
order=3,
term_raw='@{X2|sing,datv}',
definition_formal='X1'
)
data = {'original': x1.pk, 'substitution': x2.pk, 'transfer_term': True}
response = self.client.patch(
f'/api/rsforms/{self.owned.item.id}/cst-substitute',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
d1.refresh_from_db()
x2.refresh_from_db()
self.assertEqual(x2.term_raw, 'Test1')
self.assertEqual(d1.term_resolved, 'form1')
self.assertEqual(d1.definition_formal, 'X2')
def test_create_constituenta_data(self):
data = {
'alias': 'X3',
'cst_type': 'basic',
'convention': '1',
'term_raw': '2',
'definition_formal': '3',
'definition_raw': '4'
}
item = self.owned.item
response = self.client.post(
f'/api/rsforms/{item.id}/cst-create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['new_cst']['alias'], 'X3')
self.assertEqual(response.data['new_cst']['cst_type'], 'basic')
self.assertEqual(response.data['new_cst']['convention'], '1')
self.assertEqual(response.data['new_cst']['term_raw'], '2')
self.assertEqual(response.data['new_cst']['term_resolved'], '2')
self.assertEqual(response.data['new_cst']['definition_formal'], '3')
self.assertEqual(response.data['new_cst']['definition_raw'], '4')
self.assertEqual(response.data['new_cst']['definition_resolved'], '4')
def test_delete_constituenta(self):
schema = self.owned
data = {'items': [1337]}
response = self.client.patch(
f'/api/rsforms/{schema.item.id}/cst-delete-multiple',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
x1 = Constituenta.objects.create(schema=schema.item, alias='X1', cst_type='basic', order=1)
x2 = Constituenta.objects.create(schema=schema.item, alias='X2', cst_type='basic', order=2)
data = {'items': [x1.id]}
response = self.client.patch(
f'/api/rsforms/{schema.item.id}/cst-delete-multiple',
data=data, format='json'
)
x2.refresh_from_db()
schema.item.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual(len(response.data['items']), 1)
self.assertEqual(schema.constituents().count(), 1)
self.assertEqual(x2.alias, 'X2')
self.assertEqual(x2.order, 1)
x3 = Constituenta.objects.create(schema=self.unowned.item, alias='X1', cst_type='basic', order=1)
data = {'items': [x3.id]}
response = self.client.patch(
f'/api/rsforms/{schema.item.id}/cst-delete-multiple',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_move_constituenta(self):
item = self.owned.item
data = {'items': [1337], 'move_to': 1}
response = self.client.patch(
f'/api/rsforms/{item.id}/cst-moveto',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
x1 = Constituenta.objects.create(schema=item, alias='X1', cst_type='basic', order=1)
x2 = Constituenta.objects.create(schema=item, alias='X2', cst_type='basic', order=2)
data = {'items': [x2.id], 'move_to': 1}
response = self.client.patch(
f'/api/rsforms/{item.id}/cst-moveto',
data=data, format='json'
)
x1.refresh_from_db()
x2.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['id'], item.id)
self.assertEqual(x1.order, 2)
self.assertEqual(x2.order, 1)
x3 = Constituenta.objects.create(schema=self.unowned.item, alias='X1', cst_type='basic', order=1)
data = {'items': [x3.id], 'move_to': 1}
response = self.client.patch(
f'/api/rsforms/{item.id}/cst-moveto',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_reset_aliases(self):
item = self.owned.item
response = self.client.patch(f'/api/rsforms/{item.id}/reset-aliases')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['id'], item.id)
x2 = Constituenta.objects.create(schema=item, alias='X2', cst_type='basic', order=1)
x1 = Constituenta.objects.create(schema=item, alias='X1', cst_type='basic', order=2)
d11 = Constituenta.objects.create(schema=item, alias='D11', cst_type='term', order=3)
response = self.client.patch(f'/api/rsforms/{item.id}/reset-aliases')
x1.refresh_from_db()
x2.refresh_from_db()
d11.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(x2.order, 1)
self.assertEqual(x2.alias, 'X1')
self.assertEqual(x1.order, 2)
self.assertEqual(x1.alias, 'X2')
self.assertEqual(d11.order, 3)
self.assertEqual(d11.alias, 'D1')
response = self.client.patch(f'/api/rsforms/{item.id}/reset-aliases')
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_load_trs(self):
schema = self.owned
schema.item.title = 'Test11'
schema.item.save()
x1 = Constituenta.objects.create(schema=schema.item, alias='X1', cst_type='basic', order=1)
work_dir = os.path.dirname(os.path.abspath(__file__))
with open(f'{work_dir}/data/sample-rsform.trs', 'rb') as file:
data = {'file': file, 'load_metadata': False}
response = self.client.patch(
f'/api/rsforms/{schema.item.id}/load-trs',
data=data, format='multipart'
)
schema.item.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(schema.item.title, 'Test11')
self.assertEqual(len(response.data['items']), 25)
self.assertEqual(schema.constituents().count(), 25)
self.assertFalse(Constituenta.objects.filter(pk=x1.id).exists())
def test_clone(self):
item = self.owned.item
item.title = 'Test11'
item.save()
x1 = Constituenta.objects.create(schema=item, alias='X12', cst_type='basic', order=1)
d1 = Constituenta.objects.create(schema=item, alias='D2', cst_type='term', order=1)
x1.term_raw = 'человек'
x1.term_resolved = 'человек'
d1.term_raw = '@{X12|plur}'
d1.term_resolved = 'люди'
x1.save()
d1.save()
data = {'title': 'Title'}
response = self.client.post(
f'/api/library/{item.id}/clone',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['title'], 'Title')
self.assertEqual(response.data['items'][0]['alias'], x1.alias)
self.assertEqual(response.data['items'][0]['term_raw'], x1.term_raw)
self.assertEqual(response.data['items'][0]['term_resolved'], x1.term_resolved)
self.assertEqual(response.data['items'][1]['term_raw'], d1.term_raw)
self.assertEqual(response.data['items'][1]['term_resolved'], d1.term_resolved)

View File

@ -0,0 +1,109 @@
''' Testing views '''
import os
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework.exceptions import ErrorDetail
from rest_framework import status
from apps.users.models import User
from apps.rsform.models import RSForm
from apps.rsform.views import (
convert_to_ascii,
convert_to_math,
parse_expression
)
class TestRSLanguageViews(APITestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create(username='UserTest')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
def test_create_rsform(self):
work_dir = os.path.dirname(os.path.abspath(__file__))
with open(f'{work_dir}/data/sample-rsform.trs', 'rb') as file:
data = {'file': file, 'title': 'Test123', 'comment': '123', 'alias': 'ks1'}
response = self.client.post(
'/api/rsforms/create-detailed',
data=data, format='multipart'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['owner'], self.user.pk)
self.assertEqual(response.data['title'], 'Test123')
self.assertEqual(response.data['alias'], 'ks1')
self.assertEqual(response.data['comment'], '123')
def test_create_rsform_fallback(self):
data = {'title': 'Test123', 'comment': '123', 'alias': 'ks1'}
response = self.client.post(
'/api/rsforms/create-detailed',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['owner'], self.user.pk)
self.assertEqual(response.data['title'], 'Test123')
self.assertEqual(response.data['alias'], 'ks1')
self.assertEqual(response.data['comment'], '123')
def test_convert_to_ascii(self):
data = {'expression': '1=1'}
request = self.factory.post(
'/api/rslang/to-ascii',
data=data, format='json'
)
response = convert_to_ascii(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['result'], r'1 \eq 1')
def test_convert_to_ascii_missing_data(self):
data = {'data': '1=1'}
request = self.factory.post(
'/api/rslang/to-ascii',
data=data, format='json'
)
response = convert_to_ascii(request)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIsInstance(response.data['expression'][0], ErrorDetail)
def test_convert_to_math(self):
data = {'expression': r'1 \eq 1'}
request = self.factory.post(
'/api/rslang/to-math',
data=data, format='json'
)
response = convert_to_math(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['result'], r'1=1')
def test_convert_to_math_missing_data(self):
data = {'data': r'1 \eq 1'}
request = self.factory.post(
'/api/rslang/to-math',
data=data, format='json'
)
response = convert_to_math(request)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIsInstance(response.data['expression'][0], ErrorDetail)
def test_parse_expression(self):
data = {'expression': r'1=1'}
request = self.factory.post(
'/api/rslang/parse-expression',
data=data, format='json'
)
response = parse_expression(request)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['parseResult'], True)
self.assertEqual(response.data['syntax'], 'math')
self.assertEqual(response.data['astText'], '[=[1][1]]')
def test_parse_expression_missing_data(self):
data = {'data': r'1=1'}
request = self.factory.post(
'/api/rslang/parse-expression',
data=data, format='json'
)
response = parse_expression(request)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIsInstance(response.data['expression'][0], ErrorDetail)

View File

@ -0,0 +1,185 @@
''' Testing API: Versions. '''
import io
from zipfile import ZipFile
from rest_framework.test import APITestCase, APIRequestFactory, APIClient
from rest_framework import status
from apps.users.models import User
from apps.rsform.models import RSForm, Constituenta
class TestVersionViews(APITestCase):
''' Testing versioning endpoints. '''
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create(username='UserTest')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.owned = RSForm.create(title='Test', alias='T1', owner=self.user)
self.unowned = RSForm.create(title='Test2', alias='T2')
self.x1 = Constituenta.objects.create(
schema=self.owned.item,
alias='X1',
cst_type='basic',
convention='testStart',
order=1
)
def test_create_version(self):
invalid_data = {'description': 'test'}
data = {'version': '1.0.0', 'description': 'test'}
invalid_id = 1338
response = self.client.post(
f'/api/rsforms/{invalid_id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.post(
f'/api/rsforms/{self.unowned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=invalid_data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertTrue('version' in response.data)
self.assertTrue('schema' in response.data)
self.assertTrue(response.data['version'] in [v['id'] for v in response.data['schema']['versions']])
def test_retrieve_version(self):
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
version_id = response.data['version']
invalid_id = 1338
response = self.client.get(f'/api/rsforms/{invalid_id}/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{self.owned.item.id}/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{invalid_id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
response = self.client.get(f'/api/rsforms/{self.unowned.item.id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.owned.item.alias = 'NewName'
self.owned.item.save()
self.x1.alias = 'X33'
self.x1.save()
response = self.client.get(f'/api/rsforms/{self.owned.item.id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertNotEqual(response.data['alias'], self.owned.item.alias)
self.assertNotEqual(response.data['items'][0]['alias'], self.x1.alias)
self.assertEqual(response.data['version'], version_id)
def test_access_version(self):
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
version_id = response.data['version']
invalid_id = version_id + 1337
response = self.client.get(f'/api/versions/{invalid_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.client.logout()
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['version'], data['version'])
self.assertEqual(response.data['description'], data['description'])
self.assertEqual(response.data['item'], self.owned.item.id)
response = self.client.patch(
f'/api/versions/{version_id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.delete(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(user=self.user)
data = {'version': '1.1.0', 'description': 'test1'}
response = self.client.patch(
f'/api/versions/{version_id}',
data=data, format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['version'], data['version'])
self.assertEqual(response.data['description'], data['description'])
response = self.client.delete(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
response = self.client.get(f'/api/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_retrieve_version_details(self):
a1 = Constituenta.objects.create(
schema=self.owned.item,
alias='A1',
cst_type='axiom',
definition_formal='X1=X1',
order=2
)
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
version_id = response.data['version']
a1.definition_formal = 'X1=X2'
a1.save()
response = self.client.get(f'/api/rsforms/{self.owned.item.id}/versions/{version_id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
loaded_a1 = response.data['items'][1]
self.assertEqual(loaded_a1['definition_formal'], 'X1=X1')
self.assertEqual(loaded_a1['parse']['status'], 'verified')
def test_export_version(self):
invalid_id = 1338
response = self.client.get(f'/api/versions/{invalid_id}/export-file')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
data = {'version': '1.0.0', 'description': 'test'}
response = self.client.post(
f'/api/rsforms/{self.owned.item.id}/versions/create',
data=data, format='json'
)
version_id = response.data['version']
response = self.client.get(f'/api/versions/{version_id}/export-file')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.headers['Content-Disposition'],
f'attachment; filename={self.owned.item.alias}.trs'
)
with io.BytesIO(response.content) as stream:
with ZipFile(stream, 'r') as zipped_file:
self.assertIsNone(zipped_file.testzip())
self.assertIn('document.json', zipped_file.namelist())

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,7 @@
''' Utilities for testing. '''
from apps.rsform.models import LibraryItem
def response_contains(response, item: LibraryItem) -> bool:
''' Check if response contains specific item. '''
return any(x for x in response.data if x['id'] == item.pk)

View File

@ -0,0 +1,27 @@
''' REST API: Endpoint processors. '''
from .library import (
LibraryActiveView,
LibraryTemplatesView,
LibraryViewSet
)
from .constituents import ConstituentAPIView
from .versions import (
VersionAPIView,
create_version,
export_file,
retrieve_version
)
from .rsforms import (
RSFormViewSet, TrsImportView,
create_rsform
)
from .cctext import (
parse_text,
generate_lexeme,
inflect
)
from .rslang import (
convert_to_ascii,
convert_to_math,
parse_expression
)

View File

@ -0,0 +1,70 @@
''' Endpoints for cctext. '''
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.request import Request
from drf_spectacular.utils import extend_schema
from rest_framework import status as c
import cctext
from .. import serializers as s
@extend_schema(
summary='generate wordform',
tags=['NaturalLanguage'],
request=s.WordFormSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def inflect(request: Request):
''' Endpoint: Generate wordform with set grammemes. '''
serializer = s.WordFormSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
text = serializer.validated_data['text']
grams = serializer.validated_data['grams']
result = cctext.inflect(text, grams)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)
@extend_schema(
summary='all wordforms for current lexeme',
tags=['NaturalLanguage'],
request=s.TextSerializer,
responses={200: s.MultiFormSerializer},
auth=None
)
@api_view(['POST'])
def generate_lexeme(request: Request):
''' Endpoint: Generate complete set of wordforms for lexeme. '''
serializer = s.TextSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
nominal = serializer.validated_data['text']
result = cctext.generate_lexeme(nominal)
return Response(
status=c.HTTP_200_OK,
data=s.MultiFormSerializer.from_list(result)
)
@extend_schema(
summary='get likely parse grammemes',
tags=['NaturalLanguage'],
request=s.TextSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def parse_text(request: Request):
''' Endpoint: Get likely vocabulary parse. '''
serializer = s.TextSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
text = serializer.validated_data['text']
result = cctext.parse(text)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)

View File

@ -0,0 +1,23 @@
''' Endpoints for Constituenta. '''
from rest_framework import generics, permissions
from drf_spectacular.utils import extend_schema, extend_schema_view
from .. import models as m
from .. import serializers as s
from .. import utils
@extend_schema(tags=['Constituenta'])
@extend_schema_view()
class ConstituentAPIView(generics.RetrieveUpdateAPIView):
''' Endpoint: Get / Update Constituenta. '''
queryset = m.Constituenta.objects.all()
serializer_class = s.ConstituentaSerializer
def get_permissions(self):
result = super().get_permissions()
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.SchemaOwnerOrAdmin())
return result

View File

@ -0,0 +1,162 @@
''' Endpoints for library. '''
from typing import cast
from django.db import transaction
from django_filters.rest_framework import DjangoFilterBackend
from django.db.models import Q
from rest_framework import viewsets, filters, generics, permissions
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.request import Request
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import status as c
from .. import models as m
from .. import serializers as s
from .. import utils
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryActiveView(generics.ListAPIView):
''' Endpoint: Get list of library items available for active user. '''
permission_classes = (permissions.AllowAny,)
serializer_class = s.LibraryItemSerializer
def get_queryset(self):
if self.request.user.is_anonymous:
return m.LibraryItem.objects.filter(is_common=True).order_by('-time_update')
else:
user = cast(m.User, self.request.user)
# pylint: disable=unsupported-binary-operation
return m.LibraryItem.objects.filter(
Q(is_common=True) | Q(owner=user) | Q(subscription__user=user)
).distinct().order_by('-time_update')
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryTemplatesView(generics.ListAPIView):
''' Endpoint: Get list of templates. '''
permission_classes = (permissions.AllowAny,)
serializer_class = s.LibraryItemSerializer
def get_queryset(self):
template_ids = m.LibraryTemplate.objects.values_list('lib_source', flat=True)
return m.LibraryItem.objects.filter(pk__in=template_ids)
# pylint: disable=too-many-ancestors
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryViewSet(viewsets.ModelViewSet):
''' Endpoint: Library operations. '''
queryset = m.LibraryItem.objects.all()
serializer_class = s.LibraryItemSerializer
filter_backends = (DjangoFilterBackend, filters.OrderingFilter)
filterset_fields = ['item_type', 'owner', 'is_common', 'is_canonical']
ordering_fields = ('item_type', 'owner', 'title', 'time_update')
ordering = '-time_update'
def perform_create(self, serializer):
if not self.request.user.is_anonymous and 'owner' not in self.request.POST:
return serializer.save(owner=self.request.user)
else:
return serializer.save()
def get_permissions(self):
if self.action in ['update', 'destroy', 'partial_update']:
permission_list = [utils.ObjectOwnerOrAdmin]
elif self.action in ['create', 'clone', 'subscribe', 'unsubscribe']:
permission_list = [permissions.IsAuthenticated]
elif self.action in ['claim']:
permission_list = [utils.IsClaimable]
else:
permission_list = [permissions.AllowAny]
return [permission() for permission in permission_list]
def _get_item(self) -> m.LibraryItem:
return cast(m.LibraryItem, self.get_object())
@extend_schema(
summary='clone item including contents',
tags=['Library'],
request=s.LibraryItemSerializer,
responses={
c.HTTP_201_CREATED: s.RSFormParseSerializer,
c.HTTP_404_NOT_FOUND: None
}
)
@transaction.atomic
@action(detail=True, methods=['post'], url_path='clone')
def clone(self, request: Request, pk):
''' Endpoint: Create deep copy of library item. '''
serializer = s.LibraryItemSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
item = self._get_item()
if item.item_type == m.LibraryItemType.RSFORM:
schema = m.RSForm(item)
clone_data = s.RSFormTRSSerializer(schema).data
clone_data['item_type'] = item.item_type
clone_data['owner'] = self.request.user
clone_data['title'] = serializer.validated_data['title']
clone_data['alias'] = serializer.validated_data.get('alias', '')
clone_data['comment'] = serializer.validated_data.get('comment', '')
clone_data['is_common'] = serializer.validated_data.get('is_common', False)
clone_data['is_canonical'] = serializer.validated_data.get('is_canonical', False)
clone = s.RSFormTRSSerializer(data=clone_data, context={'load_meta': True})
clone.is_valid(raise_exception=True)
new_schema = clone.save()
return Response(
status=c.HTTP_201_CREATED,
data=s.RSFormParseSerializer(new_schema.item).data
)
return Response(status=c.HTTP_404_NOT_FOUND)
@extend_schema(
summary='claim item',
tags=['Library'],
request=None,
responses={c.HTTP_200_OK: s.LibraryItemSerializer}
)
@transaction.atomic
@action(detail=True, methods=['post'])
def claim(self, request: Request, pk=None):
''' Endpoint: Claim ownership of LibraryItem. '''
item = self._get_item()
if item.owner == self.request.user:
return Response(status=c.HTTP_304_NOT_MODIFIED)
else:
item.owner = cast(m.User, self.request.user)
item.save()
m.Subscription.subscribe(user=item.owner, item=item)
return Response(
status=c.HTTP_200_OK,
data=s.LibraryItemSerializer(item).data
)
@extend_schema(
summary='subscribe to item',
tags=['Library'],
request=None,
responses={c.HTTP_204_NO_CONTENT: None}
)
@action(detail=True, methods=['post'])
def subscribe(self, request: Request, pk):
''' Endpoint: Subscribe current user to item. '''
item = self._get_item()
m.Subscription.subscribe(user=cast(m.User, self.request.user), item=item)
return Response(status=c.HTTP_204_NO_CONTENT)
@extend_schema(
summary='unsubscribe from item',
tags=['Library'],
request=None,
responses={c.HTTP_204_NO_CONTENT: None},
)
@action(detail=True, methods=['delete'])
def unsubscribe(self, request: Request, pk):
''' Endpoint: Unsubscribe current user from item. '''
item = self._get_item()
m.Subscription.unsubscribe(user=cast(m.User, self.request.user), item=item)
return Response(status=c.HTTP_204_NO_CONTENT)

View File

@ -1,201 +1,19 @@
''' REST API: RSForms for conceptual schemas. '''
''' Endpoints for RSForm. '''
import json
from typing import cast, Union
from django.db import transaction
from django.http import HttpResponse
from django_filters.rest_framework import DjangoFilterBackend
from django.db.models import Q
from rest_framework import views, viewsets, filters, generics, permissions
from rest_framework.decorators import action, api_view, permission_classes
from rest_framework import views, viewsets, generics, permissions
from rest_framework.decorators import action, api_view
from rest_framework.response import Response
from rest_framework.request import Request
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import status as c
import pyconcept
import cctext
from . import models as m
from . import serializers as s
from . import utils
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryActiveView(generics.ListAPIView):
''' Endpoint: Get list of library items available for active user. '''
permission_classes = (permissions.AllowAny,)
serializer_class = s.LibraryItemSerializer
def get_queryset(self):
if self.request.user.is_anonymous:
return m.LibraryItem.objects.filter(is_common=True).order_by('-time_update')
else:
user = cast(m.User, self.request.user)
# pylint: disable=unsupported-binary-operation
return m.LibraryItem.objects.filter(
Q(is_common=True) | Q(owner=user) | Q(subscription__user=user)
).distinct().order_by('-time_update')
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryTemplatesView(generics.ListAPIView):
''' Endpoint: Get list of templates. '''
permission_classes = (permissions.AllowAny,)
serializer_class = s.LibraryItemSerializer
def get_queryset(self):
template_ids = m.LibraryTemplate.objects.values_list('lib_source', flat=True)
return m.LibraryItem.objects.filter(pk__in=template_ids)
@extend_schema(tags=['Constituenta'])
@extend_schema_view()
class ConstituentAPIView(generics.RetrieveUpdateAPIView):
''' Endpoint: Get / Update Constituenta. '''
queryset = m.Constituenta.objects.all()
serializer_class = s.ConstituentaSerializer
def get_permissions(self):
result = super().get_permissions()
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.SchemaOwnerOrAdmin())
return result
@extend_schema(tags=['Version'])
@extend_schema_view()
class VersionAPIView(generics.RetrieveUpdateDestroyAPIView):
''' Endpoint: Get / Update Constituenta. '''
queryset = m.Version.objects.all()
serializer_class = s.VersionSerializer
def get_permissions(self):
result = super().get_permissions()
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.ItemOwnerOrAdmin())
return result
# pylint: disable=too-many-ancestors
@extend_schema(tags=['Library'])
@extend_schema_view()
class LibraryViewSet(viewsets.ModelViewSet):
''' Endpoint: Library operations. '''
queryset = m.LibraryItem.objects.all()
serializer_class = s.LibraryItemSerializer
filter_backends = (DjangoFilterBackend, filters.OrderingFilter)
filterset_fields = ['item_type', 'owner', 'is_common', 'is_canonical']
ordering_fields = ('item_type', 'owner', 'title', 'time_update')
ordering = '-time_update'
def perform_create(self, serializer):
if not self.request.user.is_anonymous and 'owner' not in self.request.POST:
return serializer.save(owner=self.request.user)
else:
return serializer.save()
def get_permissions(self):
if self.action in ['update', 'destroy', 'partial_update']:
permission_list = [utils.ObjectOwnerOrAdmin]
elif self.action in ['create', 'clone', 'subscribe', 'unsubscribe']:
permission_list = [permissions.IsAuthenticated]
elif self.action in ['claim']:
permission_list = [utils.IsClaimable]
else:
permission_list = [permissions.AllowAny]
return [permission() for permission in permission_list]
def _get_item(self) -> m.LibraryItem:
return cast(m.LibraryItem, self.get_object())
@extend_schema(
summary='clone item including contents',
tags=['Library'],
request=s.LibraryItemSerializer,
responses={
c.HTTP_201_CREATED: s.RSFormParseSerializer,
c.HTTP_404_NOT_FOUND: None
}
)
@transaction.atomic
@action(detail=True, methods=['post'], url_path='clone')
def clone(self, request: Request, pk):
''' Endpoint: Create deep copy of library item. '''
serializer = s.LibraryItemSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
item = self._get_item()
if item.item_type == m.LibraryItemType.RSFORM:
schema = m.RSForm(item)
clone_data = s.RSFormTRSSerializer(schema).data
clone_data['item_type'] = item.item_type
clone_data['owner'] = self.request.user
clone_data['title'] = serializer.validated_data['title']
clone_data['alias'] = serializer.validated_data.get('alias', '')
clone_data['comment'] = serializer.validated_data.get('comment', '')
clone_data['is_common'] = serializer.validated_data.get('is_common', False)
clone_data['is_canonical'] = serializer.validated_data.get('is_canonical', False)
clone = s.RSFormTRSSerializer(data=clone_data, context={'load_meta': True})
clone.is_valid(raise_exception=True)
new_schema = clone.save()
return Response(
status=c.HTTP_201_CREATED,
data=s.RSFormParseSerializer(new_schema.item).data
)
return Response(status=c.HTTP_404_NOT_FOUND)
@extend_schema(
summary='claim item',
tags=['Library'],
request=None,
responses={c.HTTP_200_OK: s.LibraryItemSerializer}
)
@transaction.atomic
@action(detail=True, methods=['post'])
def claim(self, request: Request, pk=None):
''' Endpoint: Claim ownership of LibraryItem. '''
item = self._get_item()
if item.owner == self.request.user:
return Response(status=c.HTTP_304_NOT_MODIFIED)
else:
item.owner = cast(m.User, self.request.user)
item.save()
m.Subscription.subscribe(user=item.owner, item=item)
return Response(
status=c.HTTP_200_OK,
data=s.LibraryItemSerializer(item).data
)
@extend_schema(
summary='subscribe to item',
tags=['Library'],
request=None,
responses={c.HTTP_204_NO_CONTENT: None}
)
@action(detail=True, methods=['post'])
def subscribe(self, request: Request, pk):
''' Endpoint: Subscribe current user to item. '''
item = self._get_item()
m.Subscription.subscribe(user=cast(m.User, self.request.user), item=item)
return Response(status=c.HTTP_204_NO_CONTENT)
@extend_schema(
summary='unsubscribe from item',
tags=['Library'],
request=None,
responses={c.HTTP_204_NO_CONTENT: None},
)
@action(detail=True, methods=['delete'])
def unsubscribe(self, request: Request, pk):
''' Endpoint: Unsubscribe current user from item. '''
item = self._get_item()
m.Subscription.unsubscribe(user=cast(m.User, self.request.user), item=item)
return Response(status=c.HTTP_204_NO_CONTENT)
from .. import models as m
from .. import serializers as s
from .. import utils
@extend_schema(tags=['RSForm'])
@ -553,216 +371,3 @@ def _prepare_rsform_data(data: dict, request: Request, owner: Union[m.User, None
if 'is_canonical' in request.data:
is_canonical = request.data['is_canonical'] == 'true'
data['is_canonical'] = is_canonical
@extend_schema(
summary='save version for RSForm copying current content',
tags=['Version'],
request=s.VersionCreateSerializer,
responses={
c.HTTP_201_CREATED: s.NewVersionResponse,
c.HTTP_403_FORBIDDEN: None,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['POST'])
@permission_classes([permissions.IsAuthenticated])
def create_version(request: Request, pk_item: int):
''' Endpoint: Create new version for RSForm copying current content. '''
try:
item = m.LibraryItem.objects.get(pk=pk_item)
except m.LibraryItem.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
creator = request.user
if not creator.is_staff and creator != item.owner:
return Response(status=c.HTTP_403_FORBIDDEN)
version_input = s.VersionCreateSerializer(data=request.data)
version_input.is_valid(raise_exception=True)
data = s.RSFormSerializer(item).to_versioned_data()
result = m.RSForm(item).create_version(
version=version_input.validated_data['version'],
description=version_input.validated_data['description'],
data=data
)
return Response(
status=c.HTTP_201_CREATED,
data={
'version': result.pk,
'schema': s.RSFormParseSerializer(item).data
}
)
@extend_schema(
summary='retrieve versioned data for RSForm',
tags=['Version'],
request=None,
responses={
c.HTTP_200_OK: s.RSFormParseSerializer,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['GET'])
def retrieve_version(request: Request, pk_item: int, pk_version: int):
''' Endpoint: Retrieve version for RSForm. '''
try:
item = m.LibraryItem.objects.get(pk=pk_item)
except m.LibraryItem.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
try:
version = m.Version.objects.get(pk=pk_version)
except m.Version.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
if version.item != item:
return Response(status=c.HTTP_404_NOT_FOUND)
data = s.RSFormParseSerializer(item).from_versioned_data(version.pk, version.data)
return Response(
status=c.HTTP_200_OK,
data=data
)
@extend_schema(
summary='export versioned data as file',
tags=['Versions'],
request=None,
responses={
(c.HTTP_200_OK, 'application/zip'): bytes,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['GET'])
def export_file(request: Request, pk: int):
''' Endpoint: Download Exteor compatible file for versioned data. '''
try:
version = m.Version.objects.get(pk=pk)
except m.Version.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
data = s.RSFormTRSSerializer(m.RSForm(version.item)).from_versioned_data(version.data)
file = utils.write_zipped_json(data, utils.EXTEOR_INNER_FILENAME)
filename = utils.filename_for_schema(data['alias'])
response = HttpResponse(file, content_type='application/zip')
response['Content-Disposition'] = f'attachment; filename={filename}'
return response
@extend_schema(
summary='RS expression into Syntax Tree',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={c.HTTP_200_OK: s.ExpressionParseSerializer},
auth=None
)
@api_view(['POST'])
def parse_expression(request: Request):
''' Endpoint: Parse RS expression. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.parse_expression(expression)
return Response(
status=c.HTTP_200_OK,
data=json.loads(result)
)
@extend_schema(
summary='Unicode syntax to ASCII TeX',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={c.HTTP_200_OK: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def convert_to_ascii(request: Request):
''' Endpoint: Convert expression to ASCII syntax. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.convert_to_ascii(expression)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)
@extend_schema(
summary='ASCII TeX syntax to Unicode symbols',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def convert_to_math(request: Request):
''' Endpoint: Convert expression to MATH syntax. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.convert_to_math(expression)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)
@extend_schema(
summary='generate wordform',
tags=['NaturalLanguage'],
request=s.WordFormSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def inflect(request: Request):
''' Endpoint: Generate wordform with set grammemes. '''
serializer = s.WordFormSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
text = serializer.validated_data['text']
grams = serializer.validated_data['grams']
result = cctext.inflect(text, grams)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)
@extend_schema(
summary='all wordforms for current lexeme',
tags=['NaturalLanguage'],
request=s.TextSerializer,
responses={200: s.MultiFormSerializer},
auth=None
)
@api_view(['POST'])
def generate_lexeme(request: Request):
''' Endpoint: Generate complete set of wordforms for lexeme. '''
serializer = s.TextSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
nominal = serializer.validated_data['text']
result = cctext.generate_lexeme(nominal)
return Response(
status=c.HTTP_200_OK,
data=s.MultiFormSerializer.from_list(result)
)
@extend_schema(
summary='get likely parse grammemes',
tags=['NaturalLanguage'],
request=s.TextSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def parse_text(request: Request):
''' Endpoint: Get likely vocabulary parse. '''
serializer = s.TextSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
text = serializer.validated_data['text']
result = cctext.parse(text)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)

View File

@ -0,0 +1,70 @@
''' Endpoints pyconcept formal language parsing. '''
import json
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.request import Request
from drf_spectacular.utils import extend_schema
from rest_framework import status as c
import pyconcept
from .. import serializers as s
@extend_schema(
summary='RS expression into Syntax Tree',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={c.HTTP_200_OK: s.ExpressionParseSerializer},
auth=None
)
@api_view(['POST'])
def parse_expression(request: Request):
''' Endpoint: Parse RS expression. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.parse_expression(expression)
return Response(
status=c.HTTP_200_OK,
data=json.loads(result)
)
@extend_schema(
summary='Unicode syntax to ASCII TeX',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={c.HTTP_200_OK: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def convert_to_ascii(request: Request):
''' Endpoint: Convert expression to ASCII syntax. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.convert_to_ascii(expression)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)
@extend_schema(
summary='ASCII TeX syntax to Unicode symbols',
tags=['FormalLanguage'],
request=s.ExpressionSerializer,
responses={200: s.ResultTextResponse},
auth=None
)
@api_view(['POST'])
def convert_to_math(request: Request):
''' Endpoint: Convert expression to MATH syntax. '''
serializer = s.ExpressionSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
expression = serializer.validated_data['expression']
result = pyconcept.convert_to_math(expression)
return Response(
status=c.HTTP_200_OK,
data={'result': result}
)

View File

@ -0,0 +1,121 @@
''' Endpoints for versions. '''
from django.http import HttpResponse
from rest_framework import generics, permissions
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.request import Request
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import status as c
from .. import models as m
from .. import serializers as s
from .. import utils
@extend_schema(tags=['Version'])
@extend_schema_view()
class VersionAPIView(generics.RetrieveUpdateDestroyAPIView):
''' Endpoint: Get / Update Constituenta. '''
queryset = m.Version.objects.all()
serializer_class = s.VersionSerializer
def get_permissions(self):
result = super().get_permissions()
if self.request.method.upper() == 'GET':
result.append(permissions.AllowAny())
else:
result.append(utils.ItemOwnerOrAdmin())
return result
@extend_schema(
summary='save version for RSForm copying current content',
tags=['Version'],
request=s.VersionCreateSerializer,
responses={
c.HTTP_201_CREATED: s.NewVersionResponse,
c.HTTP_403_FORBIDDEN: None,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['POST'])
@permission_classes([permissions.IsAuthenticated])
def create_version(request: Request, pk_item: int):
''' Endpoint: Create new version for RSForm copying current content. '''
try:
item = m.LibraryItem.objects.get(pk=pk_item)
except m.LibraryItem.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
creator = request.user
if not creator.is_staff and creator != item.owner:
return Response(status=c.HTTP_403_FORBIDDEN)
version_input = s.VersionCreateSerializer(data=request.data)
version_input.is_valid(raise_exception=True)
data = s.RSFormSerializer(item).to_versioned_data()
result = m.RSForm(item).create_version(
version=version_input.validated_data['version'],
description=version_input.validated_data['description'],
data=data
)
return Response(
status=c.HTTP_201_CREATED,
data={
'version': result.pk,
'schema': s.RSFormParseSerializer(item).data
}
)
@extend_schema(
summary='retrieve versioned data for RSForm',
tags=['Version'],
request=None,
responses={
c.HTTP_200_OK: s.RSFormParseSerializer,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['GET'])
def retrieve_version(request: Request, pk_item: int, pk_version: int):
''' Endpoint: Retrieve version for RSForm. '''
try:
item = m.LibraryItem.objects.get(pk=pk_item)
except m.LibraryItem.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
try:
version = m.Version.objects.get(pk=pk_version)
except m.Version.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
if version.item != item:
return Response(status=c.HTTP_404_NOT_FOUND)
data = s.RSFormParseSerializer(item).from_versioned_data(version.pk, version.data)
return Response(
status=c.HTTP_200_OK,
data=data
)
@extend_schema(
summary='export versioned data as file',
tags=['Versions'],
request=None,
responses={
(c.HTTP_200_OK, 'application/zip'): bytes,
c.HTTP_404_NOT_FOUND: None
}
)
@api_view(['GET'])
def export_file(request: Request, pk: int):
''' Endpoint: Download Exteor compatible file for versioned data. '''
try:
version = m.Version.objects.get(pk=pk)
except m.Version.DoesNotExist:
return Response(status=c.HTTP_404_NOT_FOUND)
data = s.RSFormTRSSerializer(m.RSForm(version.item)).from_versioned_data(version.data)
file = utils.write_zipped_json(data, utils.EXTEOR_INNER_FILENAME)
filename = utils.filename_for_schema(data['alias'])
response = HttpResponse(file, content_type='application/zip')
response['Content-Disposition'] = f'attachment; filename={filename}'
return response