def upgrade()

in src/dispatch/database/revisions/tenant/versions/2021-08-09_b73416df5744.py [0:0]
88 lines of code
5 McCabe index (conditional complexity)

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "incident_role",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("role", sa.String(), nullable=True),
        sa.Column("enabled", sa.Boolean(), nullable=True),
        sa.Column("order", sa.Integer(), nullable=True),
        sa.Column("service_id", sa.Integer(), nullable=True),
        sa.Column("individual_id", sa.Integer(), nullable=True),
        sa.Column("project_id", sa.Integer(), nullable=True),
        sa.Column("updated_at", sa.DateTime(), nullable=True),
        sa.Column("created_at", sa.DateTime(), nullable=True),
        sa.ForeignKeyConstraint(
            ["individual_id"],
            ["individual_contact.id"],
        ),
        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
        sa.ForeignKeyConstraint(
            ["service_id"],
            ["service.id"],
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_table(
        "incident_role_incident_priority",
        sa.Column("incident_role_id", sa.Integer(), nullable=False),
        sa.Column("incident_priority_id", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["incident_priority_id"],
            ["incident_priority.id"],
        ),
        sa.ForeignKeyConstraint(
            ["incident_role_id"],
            ["incident_role.id"],
        ),
        sa.PrimaryKeyConstraint("incident_role_id", "incident_priority_id"),
    )
    op.create_table(
        "incident_role_tag",
        sa.Column("incident_role_id", sa.Integer(), nullable=False),
        sa.Column("tag_id", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["incident_role_id"],
            ["incident_role.id"],
        ),
        sa.ForeignKeyConstraint(
            ["tag_id"],
            ["tag.id"],
        ),
        sa.PrimaryKeyConstraint("incident_role_id", "tag_id"),
    )
    op.create_table(
        "incident_role_incident_type",
        sa.Column("incident_role_id", sa.Integer(), nullable=False),
        sa.Column("incident_type_id", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["incident_role_id"],
            ["incident_role.id"],
        ),
        sa.ForeignKeyConstraint(
            ["incident_type_id"],
            ["incident_type.id"],
        ),
        sa.PrimaryKeyConstraint("incident_role_id", "incident_type_id"),
    )

    # migrate old incident type mappings to incident roles
    bind = op.get_bind()
    session = Session(bind=bind)

    roles = defaultdict(list)

    for i_type in session.query(IncidentType).all():
        # group by types
        if i_type.commander_service_id:
            roles[(i_type.project_id, i_type.commander_service_id, "Incident Commander")].append(
                i_type
            )

        if i_type.liaison_service_id:
            roles[(i_type.project_id, i_type.liaison_service_id, "Liaison")].append(i_type)

    incident_priorities = session.query(IncidentPriority).all()

    for k, v in roles.items():
        project_id, service_id, role = k
        session.add(
            IncidentRole(
                project_id=project_id,
                incident_types=v,
                incident_priorities=incident_priorities,
                role=role,
                service_id=service_id,
            )
        )

    session.commit()