75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
|
"""replace callable/review/to_learn with status on tune_by_instrument
|
||
|
|
|
||
|
|
Revision ID: f1a2b3c4d5e6
|
||
|
|
Revises: e5f6a2b3c4d1
|
||
|
|
Create Date: 2026-06-09
|
||
|
|
"""
|
||
|
|
from alembic import op
|
||
|
|
import sqlalchemy as sa
|
||
|
|
from sqlalchemy.sql import table, column, select
|
||
|
|
|
||
|
|
revision = "f1a2b3c4d5e6"
|
||
|
|
down_revision = "e5f6a2b3c4d1"
|
||
|
|
branch_labels = None
|
||
|
|
depends_on = None
|
||
|
|
|
||
|
|
|
||
|
|
def upgrade() -> None:
|
||
|
|
# Add new status column
|
||
|
|
op.add_column(
|
||
|
|
"tune_by_instrument",
|
||
|
|
sa.Column("status", sa.String(), nullable=True),
|
||
|
|
)
|
||
|
|
|
||
|
|
# Migrate existing data (priority: callable > review > to_learn)
|
||
|
|
conn = op.get_bind()
|
||
|
|
tbi = table(
|
||
|
|
"tune_by_instrument",
|
||
|
|
column("id", sa.Integer),
|
||
|
|
column("callable", sa.Boolean),
|
||
|
|
column("review", sa.Boolean),
|
||
|
|
column("to_learn", sa.Boolean),
|
||
|
|
column("status", sa.String),
|
||
|
|
)
|
||
|
|
rows = conn.execute(select(tbi.c.id, tbi.c.callable, tbi.c.review, tbi.c.to_learn)).fetchall()
|
||
|
|
for row in rows:
|
||
|
|
id_, callable_, review_, to_learn_ = row
|
||
|
|
if callable_:
|
||
|
|
status = "callable"
|
||
|
|
elif review_:
|
||
|
|
status = "review"
|
||
|
|
elif to_learn_:
|
||
|
|
status = "to_learn"
|
||
|
|
else:
|
||
|
|
status = None
|
||
|
|
if status:
|
||
|
|
conn.execute(tbi.update().where(tbi.c.id == id_).values(status=status))
|
||
|
|
|
||
|
|
# Drop old columns
|
||
|
|
op.drop_column("tune_by_instrument", "callable")
|
||
|
|
op.drop_column("tune_by_instrument", "review")
|
||
|
|
op.drop_column("tune_by_instrument", "to_learn")
|
||
|
|
|
||
|
|
|
||
|
|
def downgrade() -> None:
|
||
|
|
op.add_column("tune_by_instrument", sa.Column("callable", sa.Boolean(), nullable=True))
|
||
|
|
op.add_column("tune_by_instrument", sa.Column("review", sa.Boolean(), nullable=True))
|
||
|
|
op.add_column("tune_by_instrument", sa.Column("to_learn", sa.Boolean(), nullable=True))
|
||
|
|
conn = op.get_bind()
|
||
|
|
tbi = table(
|
||
|
|
"tune_by_instrument",
|
||
|
|
column("id", sa.Integer),
|
||
|
|
column("callable", sa.Boolean),
|
||
|
|
column("review", sa.Boolean),
|
||
|
|
column("to_learn", sa.Boolean),
|
||
|
|
column("status", sa.String),
|
||
|
|
)
|
||
|
|
rows = conn.execute(select(tbi.c.id, tbi.c.status)).fetchall()
|
||
|
|
for id_, status in rows:
|
||
|
|
conn.execute(tbi.update().where(tbi.c.id == id_).values(
|
||
|
|
callable =(status == "callable"),
|
||
|
|
review =(status == "review"),
|
||
|
|
to_learn =(status == "to_learn"),
|
||
|
|
))
|
||
|
|
op.drop_column("tune_by_instrument", "status")
|