Method _load_on_pk_identity

lib/sqlalchemy/ext/baked.py:490–555

Load the given primary key identity from the database.

(self, session, query, primary_key_identity, **kw)

        return query._get_impl(ident, self._load_on_pk_identity)

    def _load_on_pk_identity(self, session, query, primary_key_identity, **kw):
        """Load the given primary key identity from the database."""

        mapper = query._raw_columns[0]._annotations["parententity"]

        _get_clause, _get_params = mapper._get_clause

        def setup(query):
            _lcl_get_clause = _get_clause
            q = query._clone()
            q._get_condition()
            q._order_by = None

            # None present in ident - turn those comparisons
            # into "IS NULL"
            if None in primary_key_identity:
                nones = {
                    _get_params[col].key
                    for col, value in zip(
                        mapper.primary_key, primary_key_identity
                    )
                    if value is None
                }
                _lcl_get_clause = sql_util.adapt_criterion_to_null(
                    _lcl_get_clause, nones
                )

            # TODO: can mapper._get_clause be pre-adapted?
            q._where_criteria = (
                sql_util._deep_annotate(_lcl_get_clause, {"_orm_adapt": True}),
            )

            for fn in self._post_criteria:
                q = fn(q)
            return q

        # cache the query against a key that includes
        # which positions in the primary key are NULL
        # (remember, we can map to an OUTER JOIN)
        bq = self.bq

        # add the clause we got from mapper._get_clause to the cache
        # key so that if a race causes multiple calls to _get_clause,
        # we've cached on ours
        bq = bq._clone()
        bq._cache_key += (_get_clause,)

        bq = bq.with_criteria(
            setup, tuple(elem is None for elem in primary_key_identity)
        )

        params = {
            _get_params[primary_key].key: id_val
            for id_val, primary_key in zip(
                primary_key_identity, mapper.primary_key
            )
        }