Method _setup_pairs

lib/sqlalchemy/orm/relationships.py:3149–3199

(self)

        return [(x._deannotate(), y._deannotate()) for x, y in collection]

    def _setup_pairs(self) -> None:
        sync_pairs: _MutableColumnPairs = []
        lrp: util.OrderedSet[Tuple[ColumnElement[Any], ColumnElement[Any]]] = (
            util.OrderedSet([])
        )
        secondary_sync_pairs: _MutableColumnPairs = []

        def go(
            joincond: ColumnElement[bool],
            collection: _MutableColumnPairs,
        ) -> None:
            def visit_binary(
                binary: BinaryExpression[Any],
                left: ColumnElement[Any],
                right: ColumnElement[Any],
            ) -> None:
                if (
                    "remote" in right._annotations
                    and "remote" not in left._annotations
                    and self.can_be_synced_fn(left)
                ):
                    lrp.add((left, right))
                elif (
                    "remote" in left._annotations
                    and "remote" not in right._annotations
                    and self.can_be_synced_fn(right)
                ):
                    lrp.add((right, left))
                if binary.operator is operators.eq and self.can_be_synced_fn(
                    left, right
                ):
                    if "foreign" in right._annotations:
                        collection.append((left, right))
                    elif "foreign" in left._annotations:
                        collection.append((right, left))

            visit_binary_product(visit_binary, joincond)

        for joincond, collection in [
            (self.primaryjoin, sync_pairs),
            (self.secondaryjoin, secondary_sync_pairs),
        ]:
            if joincond is None:
                continue
            go(joincond, collection)

        self.local_remote_pairs = self._deannotate_pairs(lrp)
        self.synchronize_pairs = self._deannotate_pairs(sync_pairs)
        self.secondary_synchronize_pairs = self._deannotate_pairs(
            secondary_sync_pairs
        )

    _track_overlapping_sync_targets: weakref.WeakKeyDictionary[
        ColumnElement[Any],