Load the given primary key identity from the database.
(
session: Session,
statement: Union[Select, FromStatement],
primary_key_identity: Optional[Tuple[Any, ...]],
*,
load_options: Optional[Sequence[ORMOption]] = None,
refresh_state: Optional[InstanceState[Any]] = None,
with_for_update: Optional[ForUpdateArg] = None,
only_load_props: Optional[Iterable[str]] = None,
identity_token: Optional[Any] = None,
no_autoflush: bool = False,
bind_arguments: Mapping[str, Any] = util.EMPTY_DICT,
execution_options: _ExecuteOptions = util.EMPTY_DICT,
require_pk_cols: bool = False,
is_user_refresh: bool = False,
)
| 533 | |
| 534 | |
| 535 | def _load_on_pk_identity( |
| 536 | session: Session, |
| 537 | statement: Union[Select, FromStatement], |
| 538 | primary_key_identity: Optional[Tuple[Any, ...]], |
| 539 | *, |
| 540 | load_options: Optional[Sequence[ORMOption]] = None, |
| 541 | refresh_state: Optional[InstanceState[Any]] = None, |
| 542 | with_for_update: Optional[ForUpdateArg] = None, |
| 543 | only_load_props: Optional[Iterable[str]] = None, |
| 544 | identity_token: Optional[Any] = None, |
| 545 | no_autoflush: bool = False, |
| 546 | bind_arguments: Mapping[str, Any] = util.EMPTY_DICT, |
| 547 | execution_options: _ExecuteOptions = util.EMPTY_DICT, |
| 548 | require_pk_cols: bool = False, |
| 549 | is_user_refresh: bool = False, |
| 550 | ): |
| 551 | """Load the given primary key identity from the database.""" |
| 552 | |
| 553 | query = statement |
| 554 | q = query._clone() |
| 555 | |
| 556 | assert not q._is_lambda_element |
| 557 | |
| 558 | if load_options is None: |
| 559 | load_options = QueryContext.default_load_options |
| 560 | |
| 561 | if ( |
| 562 | statement._compile_options |
| 563 | is SelectState.default_select_compile_options |
| 564 | ): |
| 565 | compile_options = _ORMCompileState.default_compile_options |
| 566 | else: |
| 567 | compile_options = statement._compile_options |
| 568 | |
| 569 | if primary_key_identity is not None: |
| 570 | mapper = query._propagate_attrs["plugin_subject"] |
| 571 | |
| 572 | _get_clause, _get_params = mapper._get_clause |
| 573 | |
| 574 | # None present in ident - turn those comparisons |
| 575 | # into "IS NULL" |
| 576 | if None in primary_key_identity: |
| 577 | nones = { |
| 578 | _get_params[col].key |
| 579 | for col, value in zip(mapper.primary_key, primary_key_identity) |
| 580 | if value is None |
| 581 | } |
| 582 | |
| 583 | _get_clause = sql_util.adapt_criterion_to_null(_get_clause, nones) |
| 584 | |
| 585 | if len(nones) == len(primary_key_identity): |
| 586 | util.warn( |
| 587 | "fully NULL primary key identity cannot load any " |
| 588 | "object. This condition may raise an error in a future " |
| 589 | "release." |
| 590 | ) |
| 591 | |
| 592 | q._where_criteria = (_get_clause,) |
no test coverage detected