Skip to content

processor_service

django_spire.knowledge.entry.version.services.processor_service

EntryVersionProcessorService

Bases: BaseDjangoModelService['EntryVersion']

Source code in django_spire/contrib/constructor/constructor.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(
        next(iter(self.__class__.__annotations__.values()))
    ).split('.')[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

publish

Source code in django_spire/knowledge/entry/version/services/processor_service.py 
18
19
20
21
def publish(self):
    self.obj.status = EntryVersionStatusChoices.PUBLISHED
    self.obj.published_datetime = localtime()
    self.obj.save()

add_update_delete_blocks

Source code in django_spire/knowledge/entry/version/services/processor_service.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def add_update_delete_blocks(self, block_data_list: list[dict]):
    from django_spire.knowledge.entry.version.block.models import EntryVersionBlock

    entry_blocks_to_add = []
    entry_blocks_to_update = []

    handled_block_ids = []

    old_entry_blocks = self.obj.blocks.active()

    old_entry_block_ids = [entry_block.id for entry_block in old_entry_blocks]

    for block_data in block_data_list:
        if block_data['id'] in old_entry_block_ids:
            entry_block: EntryVersionBlock = old_entry_blocks.get(id=block_data['id'])

            entry_block.type = block_data['type']
            entry_block.order = block_data['order']
            entry_block.update_editor_js_block_data_from_dict(block_data['data'])

            entry_blocks_to_update.append(entry_block)

            handled_block_ids.append(block_data['id'])

        else:
            entry_block = EntryVersionBlock.services.factory.create_validated_block(
                entry_version=self.obj,
                **block_data,
            )
            entry_blocks_to_add.append(entry_block)

            handled_block_ids.append(block_data['id'])

    entry_blocks_to_delete = [
        entry_block.id
        for entry_block in old_entry_blocks
        if entry_block.id not in handled_block_ids
    ]

    with transaction.atomic():
        EntryVersionBlock.objects.filter(id__in=entry_blocks_to_delete).delete()
        EntryVersionBlock.objects.bulk_update(entry_blocks_to_update, ['order', 'type', '_block_data', '_text_data'])
        EntryVersionBlock.objects.bulk_create(entry_blocks_to_add)