Function _extract_columns_from_class

lib/sqlalchemy/sql/_annotated_cols.py:248–356

(
    table_columns_cls: type[TypedColumns],
)

@util.preload_module("sqlalchemy.sql.schema")
def _extract_columns_from_class(
    table_columns_cls: type[TypedColumns],
) -> list[Column[Any]]:
    columns: dict[str, Column[Any]] = {}

    Column = util.preloaded.sql_schema.Column
    NULL_UNSPECIFIED = util.preloaded.sql_schema.NULL_UNSPECIFIED

    for base in table_columns_cls.__mro__[::-1]:
        if base in TypedColumns.__mro__:
            continue

        # _ClassScanAbstractConfig._cls_attr_resolver
        cls_annotations = util.get_annotations(base)
        cls_vars = vars(base)
        items = [
            (n, cls_vars.get(n), cls_annotations.get(n))
            for n in util.merge_lists_w_ordering(
                list(cls_vars), list(cls_annotations)
            )
            if not dunders_re.match(n)
        ]
        # --
        for name, obj, annotation in items:
            if obj is None:
                assert annotation is not None
                # no attribute, just annotation
                extracted_type = _collect_annotation(
                    table_columns_cls, name, base.__module__, annotation
                )
                if extracted_type is _NoArg.NO_ARG:
                    raise ArgumentError(
                        "No type information could be extracted from "
                        f"annotation {annotation} for attribute "
                        f"'{base.__name__}.{name}'"
                    )
                sqltype = _get_sqltype(extracted_type)
                if sqltype is None:
                    raise ArgumentError(
                        f"Could not find a SQL type for type {extracted_type} "
                        f"obtained from annotation {annotation} in "
                        f"attribute '{base.__name__}.{name}'"
                    )
                columns[name] = Column(
                    name,
                    sqltype,
                    nullable=sa_typing.includes_none(extracted_type),
                )
            elif isinstance(obj, Column):
                # has attribute attribute
                # _DeclarativeMapperConfig._produce_column_copies
                # as with orm this case is not supported
                for fk in obj.foreign_keys:
                    if (
                        fk._table_column is not None
                        and fk._table_column.table is None
                    ):
                        raise InvalidRequestError(