def upgrade()

in src/dispatch/database/revisions/tenant/versions/2024-11-04_928b725d64f6.py [0:0]

• 31 lines of code
• 11 McCabe index (conditional complexity)

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    conn = op.get_bind()
    inspector = Inspector.from_engine(conn)

    # Check if the table exists
    if "service_incident" in inspector.get_table_names():
        op.drop_table("service_incident")

    op.alter_column(
        "entity", "source", existing_type=sa.BOOLEAN(), type_=sa.String(), existing_nullable=True
    )

    indexes = inspector.get_indexes("entity")
    index_exists = any(index["name"] == "ix_entity_search_vector" for index in indexes)

    if index_exists:
        op.drop_index("ix_entity_search_vector", table_name="entity", postgresql_using="gin")

    index_exists = any(index["name"] == "entity_search_vector_idx" for index in indexes)
    if not index_exists:
        op.create_index(
            "entity_search_vector_idx",
            "entity",
            ["search_vector"],
            unique=False,
            postgresql_using="gin",
        )
    op.alter_column("entity_type", "jpath", existing_type=sa.VARCHAR(), nullable=True)

    columns = inspector.get_columns("plugin_instance")
    column_exists = any(column["name"] == "configuration" for column in columns)
    if column_exists:
        op.drop_column("plugin_instance", "configuration")

    foreign_keys = inspector.get_foreign_keys("project")

    constraint_name = "project_stable_priority_id_fkey"
    constraint_exists = any(fk["name"] == constraint_name for fk in foreign_keys)
    if constraint_exists:
        op.drop_constraint(constraint_name, "project", type_="foreignkey")