Method _reflect_fk

lib/sqlalchemy/engine/reflection.py:1729–1816

(
        self,
        _reflect_info: _ReflectionInfo,
        table_key: TableKey,
        table: sa_schema.Table,
        cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
        include_columns: Optional[Collection[str]],
        exclude_columns: Collection[str],
        resolve_fks: bool,
        _extend_on: Optional[Set[sa_schema.Table]],
        reflection_options: Dict[str, Any],
    )

            table.primary_key._reload(pk_cols)

    def _reflect_fk(
        self,
        _reflect_info: _ReflectionInfo,
        table_key: TableKey,
        table: sa_schema.Table,
        cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
        include_columns: Optional[Collection[str]],
        exclude_columns: Collection[str],
        resolve_fks: bool,
        _extend_on: Optional[Set[sa_schema.Table]],
        reflection_options: Dict[str, Any],
    ) -> None:
        fkeys = _reflect_info.foreign_keys.get(table_key, [])
        for fkey_d in fkeys:
            conname = fkey_d["name"]
            # look for columns by orig name in cols_by_orig_name,
            # but support columns that are in-Python only as fallback
            constrained_columns = [
                cols_by_orig_name[c].key if c in cols_by_orig_name else c
                for c in fkey_d["constrained_columns"]
            ]

            if (
                exclude_columns
                and set(constrained_columns).intersection(exclude_columns)
                or (
                    include_columns
                    and set(constrained_columns).difference(include_columns)
                )
            ):
                continue

            referred_schema = fkey_d["referred_schema"]
            referred_table = fkey_d["referred_table"]
            referred_columns = fkey_d["referred_columns"]
            refspec = []
            if referred_schema is not None:
                if resolve_fks:
                    sa_schema.Table(
                        referred_table,
                        table.metadata,
                        schema=referred_schema,
                        autoload_with=self.bind,
                        _extend_on=_extend_on,
                        _reflect_info=_reflect_info,
                        **reflection_options,
                    )
                for column in referred_columns:
                    refspec.append(
                        ".".join([referred_schema, referred_table, column])
                    )
            else:
                if resolve_fks:
                    sa_schema.Table(
                        referred_table,
                        table.metadata,
                        autoload_with=self.bind,
                        schema=sa_schema.BLANK_SCHEMA,