MCPcopy Create free account
hub / github.com/sqlalchemy/sqlalchemy / _get_impl

Method _get_impl

lib/sqlalchemy/orm/session.py:3764–3889  ·  view source on GitHub ↗
(
        self,
        entity: _EntityBindKey[_O],
        primary_key_identity: _PKIdentityArgument,
        db_load_fn: Callable[..., _O],
        *,
        options: Optional[Sequence[ExecutableOption]] = None,
        populate_existing: bool = False,
        with_for_update: ForUpdateParameter = None,
        identity_token: Optional[Any] = None,
        execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
        bind_arguments: Optional[_BindArguments] = None,
    )

Source from the content-addressed store, hash-verified

3762 return instance
3763
3764 def _get_impl(
3765 self,
3766 entity: _EntityBindKey[_O],
3767 primary_key_identity: _PKIdentityArgument,
3768 db_load_fn: Callable[..., _O],
3769 *,
3770 options: Optional[Sequence[ExecutableOption]] = None,
3771 populate_existing: bool = False,
3772 with_for_update: ForUpdateParameter = None,
3773 identity_token: Optional[Any] = None,
3774 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
3775 bind_arguments: Optional[_BindArguments] = None,
3776 ) -> Optional[_O]:
3777 # convert composite types to individual args
3778 if (
3779 is_composite_class(primary_key_identity)
3780 and type(primary_key_identity)
3781 in descriptor_props._composite_getters
3782 ):
3783 getter = descriptor_props._composite_getters[
3784 type(primary_key_identity)
3785 ]
3786 primary_key_identity = getter(primary_key_identity)
3787
3788 mapper: Optional[Mapper[_O]] = inspect(entity)
3789
3790 if mapper is None or not mapper.is_mapper:
3791 raise sa_exc.ArgumentError(
3792 "Expected mapped class or mapper, got: %r" % entity
3793 )
3794
3795 is_dict = isinstance(primary_key_identity, dict)
3796 if not is_dict:
3797 primary_key_identity = util.to_list(
3798 primary_key_identity, default=[None]
3799 )
3800
3801 if len(primary_key_identity) != len(mapper.primary_key):
3802 raise sa_exc.InvalidRequestError(
3803 "Incorrect number of values in identifier to formulate "
3804 "primary key for session.get(); primary key columns "
3805 "are %s" % ",".join("'%s'" % c for c in mapper.primary_key)
3806 )
3807
3808 if is_dict:
3809 pk_synonyms = mapper._pk_synonyms
3810
3811 if pk_synonyms:
3812 correct_keys = set(pk_synonyms).intersection(
3813 primary_key_identity
3814 )
3815
3816 if correct_keys:
3817 primary_key_identity = dict(primary_key_identity)
3818 for k in correct_keys:
3819 primary_key_identity[pk_synonyms[k]] = (
3820 primary_key_identity[k]
3821 )

Callers 1

getMethod · 0.95

Calls 9

_identity_lookupMethod · 0.95
inspectFunction · 0.90
is_composite_classFunction · 0.85
_from_argumentMethod · 0.80
joinMethod · 0.45
intersectionMethod · 0.45
set_label_styleMethod · 0.45
selectMethod · 0.45
optionsMethod · 0.45

Tested by

no test coverage detected