def upgrade()

in migrations/versions/640888ce445d_.py [0:0]

• 46 lines of code
• 4 McCabe index (conditional complexity)

def upgrade():
    # Copy "schedule" column into "old_schedule" column
    op.add_column(
        "queries", sa.Column("old_schedule", sa.String(length=10), nullable=True)
    )

    queries = table(
        "queries",
        sa.Column("schedule", sa.String(length=10)),
        sa.Column("old_schedule", sa.String(length=10)),
    )

    op.execute(queries.update().values({"old_schedule": queries.c.schedule}))

    # Recreate "schedule" column as a dict type
    op.drop_column("queries", "schedule")
    op.add_column(
        "queries",
        sa.Column(
            "schedule",
            MutableDict.as_mutable(PseudoJSON),
            nullable=False,
            server_default=json.dumps({}),
        ),
    )

    # Move over values from old_schedule
    queries = table(
        "queries",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("schedule", MutableDict.as_mutable(PseudoJSON)),
        sa.Column("old_schedule", sa.String(length=10)),
    )

    conn = op.get_bind()
    for query in conn.execute(queries.select()):
        schedule_json = {
            "interval": None,
            "until": None,
            "day_of_week": None,
            "time": None,
        }

        if query.old_schedule is not None:
            if ":" in query.old_schedule:
                schedule_json["interval"] = 86400
                schedule_json["time"] = query.old_schedule
            else:
                schedule_json["interval"] = int(query.old_schedule)

        conn.execute(
            queries.update()
            .where(queries.c.id == query.id)
            .values(schedule=MutableDict(schedule_json))
        )

    op.drop_column("queries", "old_schedule")