(
cls,
a: FromClause,
a_subset: Optional[FromClause],
b: FromClause,
consider_as_foreign_keys: Optional[AbstractSet[ColumnClause[Any]]],
)
| 1488 | @classmethod |
| 1489 | @util.preload_module("sqlalchemy.sql.util") |
| 1490 | def _joincond_scan_left_right( |
| 1491 | cls, |
| 1492 | a: FromClause, |
| 1493 | a_subset: Optional[FromClause], |
| 1494 | b: FromClause, |
| 1495 | consider_as_foreign_keys: Optional[AbstractSet[ColumnClause[Any]]], |
| 1496 | ) -> collections.defaultdict[ |
| 1497 | Optional[ForeignKeyConstraint], |
| 1498 | List[Tuple[ColumnClause[Any], ColumnClause[Any]]], |
| 1499 | ]: |
| 1500 | sql_util = util.preloaded.sql_util |
| 1501 | |
| 1502 | a = coercions.expect(roles.FromClauseRole, a) |
| 1503 | b = coercions.expect(roles.FromClauseRole, b) |
| 1504 | |
| 1505 | constraints: collections.defaultdict[ |
| 1506 | Optional[ForeignKeyConstraint], |
| 1507 | List[Tuple[ColumnClause[Any], ColumnClause[Any]]], |
| 1508 | ] = collections.defaultdict(list) |
| 1509 | |
| 1510 | for left in (a_subset, a): |
| 1511 | if left is None: |
| 1512 | continue |
| 1513 | for fk in sorted( |
| 1514 | b.foreign_keys, |
| 1515 | key=lambda fk: fk.parent._creation_order, |
| 1516 | ): |
| 1517 | if ( |
| 1518 | consider_as_foreign_keys is not None |
| 1519 | and fk.parent not in consider_as_foreign_keys |
| 1520 | ): |
| 1521 | continue |
| 1522 | try: |
| 1523 | col = fk.get_referent(left) |
| 1524 | except exc.NoReferenceError as nrte: |
| 1525 | table_names = {t.name for t in sql_util.find_tables(left)} |
| 1526 | if nrte.table_name in table_names: |
| 1527 | raise |
| 1528 | else: |
| 1529 | continue |
| 1530 | |
| 1531 | if col is not None: |
| 1532 | constraints[fk.constraint].append((col, fk.parent)) |
| 1533 | if left is not b: |
| 1534 | for fk in sorted( |
| 1535 | left.foreign_keys, |
| 1536 | key=lambda fk: fk.parent._creation_order, |
| 1537 | ): |
| 1538 | if ( |
| 1539 | consider_as_foreign_keys is not None |
| 1540 | and fk.parent not in consider_as_foreign_keys |
| 1541 | ): |
| 1542 | continue |
| 1543 | try: |
| 1544 | col = fk.get_referent(b) |
| 1545 | except exc.NoReferenceError as nrte: |
| 1546 | table_names = {t.name for t in sql_util.find_tables(b)} |
| 1547 | if nrte.table_name in table_names: |
no test coverage detected