(
self,
session: Session,
source_state: InstanceState[Any],
source_dict: _InstanceDict,
dest_state: InstanceState[Any],
dest_dict: _InstanceDict,
load: bool,
_recursive: Dict[Any, object],
_resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
)
| 1451 | return self._format_as_string(self.parent.class_, self.key) |
| 1452 | |
| 1453 | def merge( |
| 1454 | self, |
| 1455 | session: Session, |
| 1456 | source_state: InstanceState[Any], |
| 1457 | source_dict: _InstanceDict, |
| 1458 | dest_state: InstanceState[Any], |
| 1459 | dest_dict: _InstanceDict, |
| 1460 | load: bool, |
| 1461 | _recursive: Dict[Any, object], |
| 1462 | _resolve_conflict_map: Dict[_IdentityKeyType[Any], object], |
| 1463 | ) -> None: |
| 1464 | if load: |
| 1465 | for r in self._reverse_property: |
| 1466 | if (source_state, r) in _recursive: |
| 1467 | return |
| 1468 | |
| 1469 | if "merge" not in self._cascade: |
| 1470 | return |
| 1471 | |
| 1472 | if self.key not in source_dict: |
| 1473 | return |
| 1474 | |
| 1475 | if self.uselist: |
| 1476 | impl = source_state.get_impl(self.key) |
| 1477 | |
| 1478 | assert is_has_collection_adapter(impl) |
| 1479 | instances_iterable = impl.get_collection(source_state, source_dict) |
| 1480 | |
| 1481 | # if this is a CollectionAttributeImpl, then empty should |
| 1482 | # be False, otherwise "self.key in source_dict" should not be |
| 1483 | # True |
| 1484 | assert not instances_iterable.empty if impl.collection else True |
| 1485 | |
| 1486 | if load: |
| 1487 | # for a full merge, pre-load the destination collection, |
| 1488 | # so that individual _merge of each item pulls from identity |
| 1489 | # map for those already present. |
| 1490 | # also assumes CollectionAttributeImpl behavior of loading |
| 1491 | # "old" list in any case |
| 1492 | dest_state.get_impl(self.key).get( |
| 1493 | dest_state, dest_dict, passive=PassiveFlag.PASSIVE_MERGE |
| 1494 | ) |
| 1495 | |
| 1496 | dest_list = [] |
| 1497 | for current in instances_iterable: |
| 1498 | current_state = attributes.instance_state(current) |
| 1499 | current_dict = attributes.instance_dict(current) |
| 1500 | _recursive[(current_state, self)] = True |
| 1501 | obj = session._merge( |
| 1502 | current_state, |
| 1503 | current_dict, |
| 1504 | load=load, |
| 1505 | _recursive=_recursive, |
| 1506 | _resolve_conflict_map=_resolve_conflict_map, |
| 1507 | ) |
| 1508 | if obj is not None: |
| 1509 | dest_list.append(obj) |
| 1510 |
nothing calls this directly
no test coverage detected