from alembic import op from sqlalchemy import text from sqlalchemy.engine import reflection def table_has_column(table: str, column: str): if not hasattr(table_has_column, "inspection"): conn = op.get_bind() insp = table_has_column.inspection = reflection.Inspector.from_engine(conn) else: insp = table_has_column.inspection has_column = False for col in insp.get_columns(table): if column not in col["name"]: continue has_column = True return has_column def table_exists(table): conn = op.get_bind() inspector = reflection.Inspector.from_engine(conn) tables = inspector.get_table_names() return table in tables