111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""replace interfaces with interface_type and add requires_port
|
|
|
|
Revision ID: 0015_single_interface
|
|
Revises: 0014_merge_heads
|
|
Create Date: 2026-05-22 22:00:00.000000
|
|
|
|
"""
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy import inspect
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = "0015_single_interface"
|
|
down_revision: Union[str, Sequence[str], None] = "0014_merge_heads"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def _get_dialect() -> str:
|
|
"""Get the current database dialect name."""
|
|
conn = op.get_bind()
|
|
return conn.dialect.name
|
|
|
|
|
|
def upgrade() -> None:
|
|
dialect = _get_dialect()
|
|
|
|
# Add new columns
|
|
op.add_column('tool_types', sa.Column('interface_type', sa.String(20), nullable=True))
|
|
op.add_column('tool_types', sa.Column('requires_port', sa.Boolean(), nullable=False, server_default='true'))
|
|
|
|
# Migrate data: take first element from interfaces JSON array
|
|
if dialect == 'postgresql':
|
|
op.execute("""
|
|
UPDATE tool_types
|
|
SET interface_type = COALESCE(
|
|
(SELECT elem FROM jsonb_array_elements_text(interfaces::jsonb) AS elem LIMIT 1),
|
|
'web'
|
|
),
|
|
requires_port = CASE
|
|
WHEN COALESCE(
|
|
(SELECT elem FROM jsonb_array_elements_text(interfaces::jsonb) AS elem LIMIT 1),
|
|
'web'
|
|
) = 'web' THEN true
|
|
ELSE false
|
|
END
|
|
""")
|
|
else:
|
|
# SQLite: interfaces is stored as JSON text, extract first array element
|
|
op.execute("""
|
|
UPDATE tool_types
|
|
SET interface_type = COALESCE(
|
|
(SELECT json_extract(value, '$[0]')
|
|
FROM json_each(interfaces) AS value
|
|
WHERE json_valid(interfaces)
|
|
LIMIT 1),
|
|
'web'
|
|
),
|
|
requires_port = CASE
|
|
WHEN COALESCE(
|
|
(SELECT json_extract(value, '$[0]')
|
|
FROM json_each(interfaces) AS value
|
|
WHERE json_valid(interfaces)
|
|
LIMIT 1),
|
|
'web'
|
|
) = 'web' THEN true
|
|
ELSE false
|
|
END
|
|
""")
|
|
|
|
# Make interface_type non-nullable after data migration
|
|
op.alter_column('tool_types', 'interface_type', nullable=False)
|
|
|
|
# Drop old interfaces column
|
|
op.drop_column('tool_types', 'interfaces')
|
|
|
|
# Add CHECK constraint for interface_type (only on PostgreSQL; SQLite supports it too)
|
|
op.create_check_constraint('chk_interface_type', 'tool_types', sa.text("interface_type IN ('web', 'terminal')"))
|
|
|
|
|
|
def downgrade() -> None:
|
|
dialect = _get_dialect()
|
|
|
|
# Drop CHECK constraint
|
|
op.drop_constraint('chk_interface_type', 'tool_types', type_='check')
|
|
|
|
# Add back interfaces column
|
|
if dialect == 'postgresql':
|
|
op.add_column('tool_types', sa.Column('interfaces', postgresql.JSONB(astext_type=sa.Text()), nullable=False, server_default='["web"]'))
|
|
|
|
# Migrate data back: wrap interface_type in array
|
|
op.execute("""
|
|
UPDATE tool_types
|
|
SET interfaces = jsonb_build_array(interface_type)
|
|
""")
|
|
else:
|
|
op.add_column('tool_types', sa.Column('interfaces', sa.JSON(), nullable=False, server_default='["web"]'))
|
|
|
|
# Migrate data back: wrap interface_type in array for SQLite
|
|
op.execute("""
|
|
UPDATE tool_types
|
|
SET interfaces = json_array(interface_type)
|
|
""")
|
|
|
|
# Drop new columns
|
|
op.drop_column('tool_types', 'requires_port')
|
|
op.drop_column('tool_types', 'interface_type')
|