Method create_row_processor

lib/sqlalchemy/orm/util.py:1711–1754

Produce the "row processing" function for this :class:`.Bundle`. May be overridden by subclasses to provide custom behaviors when results are fetched. The method is passed the statement object and a set of "row processor" functions at query execution time; these proc

(
        self,
        query: Select[Any],
        procs: Sequence[Callable[[Row[Any]], Any]],
        labels: Sequence[str],
    )

        return cloned

    def create_row_processor(
        self,
        query: Select[Any],
        procs: Sequence[Callable[[Row[Any]], Any]],
        labels: Sequence[str],
    ) -> Callable[[Row[Any]], Any]:
        """Produce the "row processing" function for this :class:`.Bundle`.

        May be overridden by subclasses to provide custom behaviors when
        results are fetched. The method is passed the statement object and a
        set of "row processor" functions at query execution time; these
        processor functions when given a result row will return the individual
        attribute value, which can then be adapted into any kind of return data
        structure.

        The example below illustrates replacing the usual :class:`.Row`
        return structure with a straight Python dictionary::

            from sqlalchemy.orm import Bundle


            class DictBundle(Bundle):
                def create_row_processor(self, query, procs, labels):
                    "Override create_row_processor to return values as dictionaries"

                    def proc(row):
                        return dict(zip(labels, (proc(row) for proc in procs)))

                    return proc

        A result from the above :class:`_orm.Bundle` will return dictionary
        values::

            bn = DictBundle("mybundle", MyClass.data1, MyClass.data2)
            for row in session.execute(select(bn)).where(bn.c.data1 == "d1"):
                print(row.mybundle["data1"], row.mybundle["data2"])

        """  # noqa: E501
        keyed_tuple = result_tuple(labels, [() for l in labels])

        def proc(row: Row[Any]) -> Any:
            return keyed_tuple([proc(row) for proc in procs])

        return proc


def _orm_annotate(element: _SA, exclude: Optional[Any] = None) -> _SA: