Method cascade_iterator

lib/sqlalchemy/orm/mapper.py:3929–4024

r"""Iterate each element and its mapper in an object graph, for all relationships that meet the given cascade rule. :param type\_: The name of the cascade rule (i.e. ``"save-update"``, ``"delete"``, etc.). .. note:: the ``"all"`` cascade is not accept

(
        self,
        type_: str,
        state: InstanceState[_O],
        halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
    )

        return self._subclass_load_via_in(self, self.base_mapper)

    def cascade_iterator(
        self,
        type_: str,
        state: InstanceState[_O],
        halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
    ) -> Iterator[
        Tuple[object, Mapper[Any], InstanceState[Any], _InstanceDict]
    ]:
        r"""Iterate each element and its mapper in an object graph,
        for all relationships that meet the given cascade rule.

        :param type\_:
          The name of the cascade rule (i.e. ``"save-update"``, ``"delete"``,
          etc.).

          .. note::  the ``"all"`` cascade is not accepted here.  For a generic
             object traversal function, see :ref:`faq_walk_objects`.

        :param state:
          The lead InstanceState.  child items will be processed per
          the relationships defined for this object's mapper.

        :return: the method yields individual object instances.

        .. seealso::

            :ref:`unitofwork_cascades`

            :ref:`faq_walk_objects` - illustrates a generic function to
            traverse all objects without relying on cascades.

        """
        visited_states: Set[InstanceState[Any]] = set()
        prp, mpp = object(), object()

        assert state.mapper.isa(self)

        # this is actually a recursive structure, fully typing it seems
        # a little too difficult for what it's worth here
        visitables: Deque[
            Tuple[
                Deque[Any],
                object,
                Optional[InstanceState[Any]],
                Optional[_InstanceDict],
            ]
        ]

        visitables = deque(
            [(deque(state.mapper._props.values()), prp, state, state.dict)]
        )

        while visitables:
            iterator, item_type, parent_state, parent_dict = visitables[-1]
            if not iterator:
                visitables.pop()
                continue