ConceptPortal-public/rsconcept/backend/apps/oss/models/PropagationFacade.py
Ivan 9170692f33
Some checks are pending
Backend CI / build (3.12) (push) Waiting to run
R: Improve cst creation propagation
2024-08-11 21:23:45 +03:00

43 lines
1.7 KiB
Python

''' Models: Change propagation facade - managing all changes in OSS. '''
from apps.library.models import LibraryItem
from apps.rsform.models import Constituenta, RSForm
from .ChangeManager import ChangeManager
def _get_oss_hosts(item: LibraryItem) -> list[LibraryItem]:
''' Get all hosts for LibraryItem. '''
return list(LibraryItem.objects.filter(operations__result=item).only('pk'))
class PropagationFacade:
''' Change propagation API. '''
@classmethod
def after_create_cst(cls, new_cst: list[Constituenta], source: RSForm) -> None:
''' Trigger cascade resolutions when new constituent is created. '''
hosts = _get_oss_hosts(source.model)
for host in hosts:
ChangeManager(host).after_create_cst(new_cst, source)
@classmethod
def after_change_cst_type(cls, target: Constituenta, source: RSForm) -> None:
''' Trigger cascade resolutions when constituenta type is changed. '''
hosts = _get_oss_hosts(source.model)
for host in hosts:
ChangeManager(host).after_change_cst_type(target, source)
@classmethod
def after_update_cst(cls, target: Constituenta, data: dict, old_data: dict, source: RSForm) -> None:
''' Trigger cascade resolutions when constituenta data is changed. '''
hosts = _get_oss_hosts(source.model)
for host in hosts:
ChangeManager(host).after_update_cst(target, data, old_data, source)
@classmethod
def before_delete(cls, target: list[Constituenta], source: RSForm) -> None:
''' Trigger cascade resolutions before constituents are deleted. '''
hosts = _get_oss_hosts(source.model)
for host in hosts:
ChangeManager(host).before_delete(target, source)