| 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 |