openslides_backend/action/generics/delete.py
def base_update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
        """
        Takes care of on_delete handling.
        """
        # TODO: Check if instance exists in DB and is not deleted. Ensure that meta_deleted field is added to locked_fields.

        # Update instance (by default this does nothing)
        instance = self.update_instance(instance)

        # Fetch db instance with all relevant fields
        this_fqid = FullQualifiedId(self.model.collection, instance["id"])
        relevant_fields = [
            field.get_own_field_name()
            for field in self.model.get_relation_fields()
            if field.on_delete != OnDelete.SET_NULL
        ] + ["meta_deleted"]
        db_instance = self.datastore.fetch_model(
            fqid=this_fqid,
            mapped_fields=relevant_fields,
        )

        # Update instance and set relation fields to None.
        # Gather all delete actions with action data and also all models to be deleted
        delete_actions: List[Tuple[Type[Action], ActionData]] = []
        self.datastore.update_additional_models(this_fqid, DeletedModel())
        for field in self.model.get_relation_fields():
            # Check on_delete.
            if field.on_delete != OnDelete.SET_NULL:
                # Extract all foreign keys as fqids from the model
                foreign_fqids: List[FullQualifiedId] = []
                if isinstance(field, BaseTemplateRelationField):
                    structured_fields = list(
                        self.get_all_structured_fields(field, instance["id"])
                    )
                    db_instance_structured_fields = self.datastore.fetch_model(
                        this_fqid, structured_fields
                    )
                    for structured_field_name in structured_fields:
                        foreign_fqids += transform_to_fqids(
                            db_instance_structured_fields[structured_field_name],
                            field.get_target_collection(),
                        )
                else:
                    value = db_instance.get(field.get_own_field_name(), [])
                    foreign_fqids = transform_to_fqids(
                        value, field.get_target_collection()
                    )

                if field.on_delete == OnDelete.PROTECT:
                    protected_fqids = [
                        fqid for fqid in foreign_fqids if not self.is_deleted(fqid)
                    ]
                    if protected_fqids:
                        raise ProtectedModelsException(this_fqid, protected_fqids)
                else:
                    # field.on_delete == OnDelete.CASCADE
                    # Execute the delete action for all fqids
                    for fqid in foreign_fqids:
                        if self.is_deleted(fqid):
                            # skip models that are already deleted
                            continue
                        delete_action_class = actions_map.get(
                            f"{str(fqid.collection)}.delete"
                        )
                        if not delete_action_class:
                            raise ActionException(
                                f"Can't cascade the delete action to {str(fqid.collection)} "
                                "since no delete action was found."
                            )
                        # Assume that the delete action uses the standard action data
                        action_data = [{"id": fqid.id}]
                        delete_actions.append((delete_action_class, action_data))
                        self.datastore.update_additional_models(fqid, DeletedModel())
            else:
                # field.on_delete == OnDelete.SET_NULL
                if not isinstance(field, BaseTemplateRelationField):
                    fields = [field.get_own_field_name()]
                else:
                    fields = list(self.get_all_structured_fields(field, instance["id"]))

                for field_name in fields:
                    instance[field_name] = None

        # Add additional relation models and execute all previously gathered delete actions
        # catch all protected models exception to gather all protected fqids
        all_protected_fqids: List[FullQualifiedId] = []
        for delete_action_class, delete_action_data in delete_actions:
            try:
                self.execute_other_action(delete_action_class, delete_action_data)
            except ProtectedModelsException as e:
                all_protected_fqids.extend(e.fqids)

        if all_protected_fqids:
            raise ProtectedModelsException(this_fqid, all_protected_fqids)

        return instance

        Takes care of on_delete handling.
        
Takes care of on_delete handling.
Takes care of on_delete handling .
def base_update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
        instance = self.update_instance(instance)
        this_fqid = FullQualifiedId(self.model.collection, instance["id"])
        relevant_fields = [
            field.get_own_field_name()
            for field in self.model.get_relation_fields()
            if field.on_delete != OnDelete.SET_NULL
        ] + ["meta_deleted"]
        db_instance = self.datastore.fetch_model(
            fqid=this_fqid,
            mapped_fields=relevant_fields,
        )
        delete_actions: List[Tuple[Type[Action], ActionData]] = []
        self.datastore.update_additional_models(this_fqid, DeletedModel())
        for field in self.model.get_relation_fields():
            if field.on_delete != OnDelete.SET_NULL:
                foreign_fqids: List[FullQualifiedId] = []
                if isinstance(field, BaseTemplateRelationField):
                    structured_fields = list(
                        self.get_all_structured_fields(field, instance["id"])
                    )
                    db_instance_structured_fields = self.datastore.fetch_model(
                        this_fqid, structured_fields
                    )
                    for structured_field_name in structured_fields:
                        foreign_fqids += transform_to_fqids(
                            db_instance_structured_fields[structured_field_name],
                            field.get_target_collection(),
                        )
                else:
                    value = db_instance.get(field.get_own_field_name(), [])
                    foreign_fqids = transform_to_fqids(
                        value, field.get_target_collection()
                    )
                if field.on_delete == OnDelete.PROTECT:
                    protected_fqids = [
                        fqid for fqid in foreign_fqids if not self.is_deleted(fqid)
                    ]
                    if protected_fqids:
                        raise ProtectedModelsException(this_fqid, protected_fqids)
                else:
                    for fqid in foreign_fqids:
                        if self.is_deleted(fqid):
                            continue
                        delete_action_class = actions_map.get(
                            f"{str(fqid.collection)}.delete"
                        )
                        if not delete_action_class:
                            raise ActionException(
                                f"Can't cascade the delete action to {str(fqid.collection)} "
                                "since no delete action was found."
                            )
                        action_data = [{"id": fqid.id}]
                        delete_actions.append((delete_action_class, action_data))
                        self.datastore.update_additional_models(fqid, DeletedModel())
            else:
                if not isinstance(field, BaseTemplateRelationField):
                    fields = [field.get_own_field_name()]
                else:
                    fields = list(self.get_all_structured_fields(field, instance["id"]))
                for field_name in fields:
                    instance[field_name] = None
        all_protected_fqids: List[FullQualifiedId] = []
        for delete_action_class, delete_action_data in delete_actions:
            try:
                self.execute_other_action(delete_action_class, delete_action_data)
            except ProtectedModelsException as e:
                all_protected_fqids.extend(e.fqids)
        if all_protected_fqids:
            raise ProtectedModelsException(this_fqid, all_protected_fqids)
        return instance
def base_update_instance ( self , instance : Dict [ str , Any ] ) -> Dict [ str , Any ] : instance = self . update_instance ( instance ) this_fqid = FullQualifiedId ( self . model . collection , instance [ "id" ] ) relevant_fields = [ field . get_own_field_name ( ) for field in self . model . get_relation_fields ( ) if field . on_delete != OnDelete . SET_NULL ] + [ "meta_deleted" ] db_instance = self . datastore . fetch_model ( fqid = this_fqid , mapped_fields = relevant_fields , ) delete_actions : List [ Tuple [ Type [ Action ] , ActionData ] ] = [ ] self . datastore . update_additional_models ( this_fqid , DeletedModel ( ) ) for field in self . model . get_relation_fields ( ) : if field . on_delete != OnDelete . SET_NULL : foreign_fqids : List [ FullQualifiedId ] = [ ] if isinstance ( field , BaseTemplateRelationField ) : structured_fields = list ( self . get_all_structured_fields ( field , instance [ "id" ] ) ) db_instance_structured_fields = self . datastore . fetch_model ( this_fqid , structured_fields ) for structured_field_name in structured_fields : foreign_fqids += transform_to_fqids ( db_instance_structured_fields [ structured_field_name ] , field . get_target_collection ( ) , ) else : value = db_instance . get ( field . get_own_field_name ( ) , [ ] ) foreign_fqids = transform_to_fqids ( value , field . get_target_collection ( ) ) if field . on_delete == OnDelete . PROTECT : protected_fqids = [ fqid for fqid in foreign_fqids if not self . is_deleted ( fqid ) ] if protected_fqids : raise ProtectedModelsException ( this_fqid , protected_fqids ) else : for fqid in foreign_fqids : if self . is_deleted ( fqid ) : continue delete_action_class = actions_map . get ( f"{str(fqid.collection)}.delete" ) if not delete_action_class : raise ActionException ( f"Can't cascade the delete action to {str(fqid.collection)} " "since no delete action was found." ) action_data = [ { "id" : fqid . id } ] delete_actions . append ( ( delete_action_class , action_data ) ) self . datastore . update_additional_models ( fqid , DeletedModel ( ) ) else : if not isinstance ( field , BaseTemplateRelationField ) : fields = [ field . get_own_field_name ( ) ] else : fields = list ( self . get_all_structured_fields ( field , instance [ "id" ] ) ) for field_name in fields : instance [ field_name ] = None all_protected_fqids : List [ FullQualifiedId ] = [ ] for delete_action_class , delete_action_data in delete_actions : try : self . execute_other_action ( delete_action_class , delete_action_data ) except ProtectedModelsException as e : all_protected_fqids . extend ( e . fqids ) if all_protected_fqids : raise ProtectedModelsException ( this_fqid , all_protected_fqids ) return instance
Takes care of on_delete handling.
Takes care of on_delete handling .
"""
        Takes care of on_delete handling.
        """ # TODO: Check if instance exists in DB and is not deleted. Ensure that meta_deleted field is added to locked_fields. # Update instance (by default this does nothing) # Fetch db instance with all relevant fields # Update instance and set relation fields to None. # Gather all delete actions with action data and also all models to be deleted # Check on_delete. # Extract all foreign keys as fqids from the model # field.on_delete == OnDelete.CASCADE # Execute the delete action for all fqids # skip models that are already deleted # Assume that the delete action uses the standard action data # field.on_delete == OnDelete.SET_NULL # Add additional relation models and execute all previously gathered delete actions # catch all protected models exception to gather all protected fqids
self
instance
Dict[str, Any]