Files
portfolio-hub/migrations/versions/20260403_0007_bot_principal_permission_only.py
T

63 lines
1.9 KiB
Python

"""remove bot owner link and enforce bot permission
Revision ID: 20260403_0007
Revises: 20260403_0006
Create Date: 2026-04-03 14:20:00
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260403_0007"
down_revision = "20260403_0006"
branch_labels = None
depends_on = None
def _table_exists(inspector, table_name: str) -> bool:
return table_name in inspector.get_table_names()
def _column_exists(inspector, table_name: str, column_name: str) -> bool:
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
def upgrade() -> None:
bind = op.get_bind()
inspector = inspect(bind)
if not _table_exists(inspector, "bot_principals"):
return
if not _column_exists(inspector, "bot_principals", "permission"):
op.add_column("bot_principals", sa.Column("permission", sa.Text(), nullable=True))
op.execute("UPDATE bot_principals SET permission = 'read' WHERE permission IS NULL")
op.alter_column("bot_principals", "permission", existing_type=sa.Text(), nullable=False)
inspector = inspect(bind)
if _column_exists(inspector, "bot_principals", "owner_user_id"):
op.drop_column("bot_principals", "owner_user_id")
def downgrade() -> None:
bind = op.get_bind()
inspector = inspect(bind)
if not _table_exists(inspector, "bot_principals"):
return
if not _column_exists(inspector, "bot_principals", "owner_user_id"):
op.add_column("bot_principals", sa.Column("owner_user_id", sa.Text(), nullable=True))
op.execute("UPDATE bot_principals SET owner_user_id = 'legacy-owner' WHERE owner_user_id IS NULL")
op.alter_column("bot_principals", "owner_user_id", existing_type=sa.Text(), nullable=False)
inspector = inspect(bind)
if _column_exists(inspector, "bot_principals", "permission"):
op.drop_column("bot_principals", "permission")