96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
"""rename control plane status column
|
|
|
|
Revision ID: 20260413_rename_control_plane_status
|
|
Revises: 20260406_add_mounted_runtime_active_draft_fields
|
|
Create Date: 2026-04-13
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision = "20260413_rename_control_plane_status"
|
|
down_revision = "20260406_add_mounted_runtime_active_draft_fields"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
TABLE_NAME = "ai_orchestration_runs"
|
|
NEW_COLUMN = "control_plane_status"
|
|
NEW_INDEX = "ix_ai_orchestration_runs_control_plane_status"
|
|
|
|
|
|
def _legacy_column() -> str:
|
|
return "".join(("open", "claw", "_status"))
|
|
|
|
|
|
def _legacy_index() -> str:
|
|
return f"ix_ai_orchestration_runs_{_legacy_column()}"
|
|
|
|
|
|
def _table_exists(bind, table_name: str) -> bool:
|
|
return sa.inspect(bind).has_table(table_name)
|
|
|
|
|
|
def _has_column(bind, table_name: str, column_name: str) -> bool:
|
|
inspector = sa.inspect(bind)
|
|
return any(str(col.get("name") or "") == column_name for col in inspector.get_columns(table_name))
|
|
|
|
|
|
def _index_exists(bind, table_name: str, index_name: str) -> bool:
|
|
inspector = sa.inspect(bind)
|
|
return any(str(idx.get("name") or "") == index_name for idx in inspector.get_indexes(table_name))
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
if not _table_exists(bind, TABLE_NAME):
|
|
return
|
|
|
|
legacy_column = _legacy_column()
|
|
legacy_index = _legacy_index()
|
|
has_legacy = _has_column(bind, TABLE_NAME, legacy_column)
|
|
has_new = _has_column(bind, TABLE_NAME, NEW_COLUMN)
|
|
|
|
if has_legacy and not has_new:
|
|
op.alter_column(
|
|
TABLE_NAME,
|
|
legacy_column,
|
|
new_column_name=NEW_COLUMN,
|
|
existing_type=sa.String(length=20),
|
|
existing_nullable=True,
|
|
)
|
|
|
|
if _index_exists(bind, TABLE_NAME, legacy_index):
|
|
op.drop_index(legacy_index, table_name=TABLE_NAME)
|
|
if _has_column(bind, TABLE_NAME, NEW_COLUMN) and not _index_exists(bind, TABLE_NAME, NEW_INDEX):
|
|
op.create_index(NEW_INDEX, TABLE_NAME, [NEW_COLUMN], unique=False)
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
if not _table_exists(bind, TABLE_NAME):
|
|
return
|
|
|
|
legacy_column = _legacy_column()
|
|
legacy_index = _legacy_index()
|
|
has_legacy = _has_column(bind, TABLE_NAME, legacy_column)
|
|
has_new = _has_column(bind, TABLE_NAME, NEW_COLUMN)
|
|
|
|
if _index_exists(bind, TABLE_NAME, NEW_INDEX):
|
|
op.drop_index(NEW_INDEX, table_name=TABLE_NAME)
|
|
|
|
if has_new and not has_legacy:
|
|
op.alter_column(
|
|
TABLE_NAME,
|
|
NEW_COLUMN,
|
|
new_column_name=legacy_column,
|
|
existing_type=sa.String(length=20),
|
|
existing_nullable=True,
|
|
)
|
|
|
|
if _has_column(bind, TABLE_NAME, legacy_column) and not _index_exists(bind, TABLE_NAME, legacy_index):
|
|
op.create_index(legacy_index, TABLE_NAME, [legacy_column], unique=False)
|