"""phase32 — add ios_version and last_seen_at to api_device_tokens phase31 recorded as applied but ALTER statements never executed. Uses INFORMATION_SCHEMA column-existence checks — safe on MySQL 5.7+. Also drops the orphaned device_registrations table from phase30 if present. """ revision = 'phase32_device_token_columns' down_revision = 'phase31_device_registry' branch_labels = None depends_on = None from alembic import op import sqlalchemy as sa def _column_exists(conn, table, column): result = conn.execute(sa.text( "SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": table, "c": column}) return result.scalar() > 0 def _table_exists(conn, table): result = conn.execute(sa.text( "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t" ), {"t": table}) return result.scalar() > 0 def upgrade(): bind = op.get_bind() if not _column_exists(bind, 'api_device_tokens', 'ios_version'): op.execute(sa.text( "ALTER TABLE api_device_tokens ADD COLUMN ios_version VARCHAR(20) NULL" )) if not _column_exists(bind, 'api_device_tokens', 'last_seen_at'): op.execute(sa.text( "ALTER TABLE api_device_tokens ADD COLUMN last_seen_at DATETIME NULL" )) op.execute(sa.text( "UPDATE api_device_tokens SET last_seen_at = registered_at WHERE last_seen_at IS NULL" )) if _table_exists(bind, 'device_registrations'): op.execute(sa.text("DROP TABLE device_registrations")) def downgrade(): bind = op.get_bind() if _column_exists(bind, 'api_device_tokens', 'ios_version'): op.execute(sa.text( "ALTER TABLE api_device_tokens DROP COLUMN ios_version" )) if _column_exists(bind, 'api_device_tokens', 'last_seen_at'): op.execute(sa.text( "ALTER TABLE api_device_tokens DROP COLUMN last_seen_at" ))