Make initial migration downgrade resilient to missing tables
All checks were successful
ci / test (push) Successful in 17s
ci / publish (push) Successful in 54s

This commit is contained in:
Space-Banane
2026-05-22 19:36:25 +02:00
parent 7dec819a3b
commit 25d5e48dac

View File

@@ -7,8 +7,8 @@ Create Date: 2026-05-22 19:00:00
from typing import Sequence, Union from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa import sqlalchemy as sa
from alembic import op
revision: str = "0001_initial" revision: str = "0001_initial"
@@ -92,16 +92,32 @@ def upgrade() -> None:
def downgrade() -> None: def downgrade() -> None:
op.drop_index("ix_bot_comments_repo_pr", table_name="bot_comments") bind = op.get_bind()
op.drop_table("bot_comments") inspector = sa.inspect(bind)
op.drop_index("ix_review_runs_job_status", table_name="review_runs") def has_table(table_name: str) -> bool:
op.drop_table("review_runs") return table_name in inspector.get_table_names()
op.drop_index("ix_review_jobs_lookup", table_name="review_jobs") def has_index(table_name: str, index_name: str) -> bool:
op.drop_table("review_jobs") return any(index["name"] == index_name for index in inspector.get_indexes(table_name))
op.drop_table("webhook_events") if has_table("bot_comments"):
if has_index("bot_comments", "ix_bot_comments_repo_pr"):
op.drop_index("ix_bot_comments_repo_pr", table_name="bot_comments")
op.drop_table("bot_comments")
if has_table("review_runs"):
if has_index("review_runs", "ix_review_runs_job_status"):
op.drop_index("ix_review_runs_job_status", table_name="review_runs")
op.drop_table("review_runs")
if has_table("review_jobs"):
if has_index("review_jobs", "ix_review_jobs_lookup"):
op.drop_index("ix_review_jobs_lookup", table_name="review_jobs")
op.drop_table("review_jobs")
if has_table("webhook_events"):
op.drop_table("webhook_events")
sa.Enum(name="runstatus").drop(op.get_bind(), checkfirst=True) sa.Enum(name="runstatus").drop(op.get_bind(), checkfirst=True)
sa.Enum(name="jobstatus").drop(op.get_bind(), checkfirst=True) sa.Enum(name="jobstatus").drop(op.get_bind(), checkfirst=True)