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

Method _joincond_scan_left_right

lib/sqlalchemy/sql/selectable.py:1490–1556  ·  view source on GitHub ↗
(
        cls,
        a: FromClause,
        a_subset: Optional[FromClause],
        b: FromClause,
        consider_as_foreign_keys: Optional[AbstractSet[ColumnClause[Any]]],
    )

Source from the content-addressed store, hash-verified

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:

Callers 2

_join_conditionMethod · 0.80
_can_joinMethod · 0.80

Calls 2

get_referentMethod · 0.80
appendMethod · 0.45

Tested by

no test coverage detected