def update()

in src/dispatch/incident/service.py [0:0]

• 82 lines of code
• 14 McCabe index (conditional complexity)

def update(*, db_session: Session, incident: Incident, incident_in: IncidentUpdate) -> Incident:
    """Updates an existing incident."""
    incident_type = incident_type_service.get_by_name_or_default(
        db_session=db_session,
        project_id=incident.project.id,
        incident_type_in=incident_in.incident_type,
    )

    incident_severity = incident_severity_service.get_by_name_or_default(
        db_session=db_session,
        project_id=incident.project.id,
        incident_severity_in=incident_in.incident_severity,
    )

    if incident_in.status == IncidentStatus.stable and incident.project.stable_priority:
        incident_priority = incident.project.stable_priority
    else:
        incident_priority = incident_priority_service.get_by_name_or_default(
            db_session=db_session,
            project_id=incident.project.id,
            incident_priority_in=incident_in.incident_priority,
        )

    cases = []
    for c in incident_in.cases:
        cases.append(case_service.get(db_session=db_session, case_id=c.id))

    tags = []
    for t in incident_in.tags:
        tags.append(tag_service.get_or_create(db_session=db_session, tag_in=t))

    terms = []
    for t in incident_in.terms:
        terms.append(term_service.get_or_create(db_session=db_session, term_in=t))

    duplicates = []
    for d in incident_in.duplicates:
        duplicates.append(get(db_session=db_session, incident_id=d.id))

    incident_costs = []
    for incident_cost in incident_in.incident_costs:
        incident_costs.append(
            incident_cost_service.get_or_create(
                db_session=db_session, incident_cost_in=incident_cost
            )
        )

    # Update total incident response cost if incident type has changed.
    if incident_type.id != incident.incident_type.id:
        incident_cost_service.update_incident_response_cost(
            incident_id=incident.id, db_session=db_session
        )
        # if the new incident type has plugin metadata and the
        # project key of the ticket is the same, also update the ticket with the new metadata
        if incident_type.plugin_metadata:
            ticket_flows.update_incident_ticket_metadata(
                db_session=db_session,
                ticket_id=incident.ticket.resource_id,
                project_id=incident.project.id,
                incident_id=incident.id,
                incident_type=incident_type,
            )

    update_data = incident_in.dict(
        skip_defaults=True,
        exclude={
            "cases",
            "commander",
            "duplicates",
            "incident_costs",
            "incident_priority",
            "incident_severity",
            "incident_type",
            "project",
            "reporter",
            "status",
            "tags",
            "terms",
            "visibility",
        },
    )

    for field in update_data.keys():
        setattr(incident, field, update_data[field])

    incident.cases = cases
    incident.duplicates = duplicates
    incident.incident_costs = incident_costs
    incident.incident_priority = incident_priority
    incident.incident_severity = incident_severity
    incident.incident_type = incident_type
    incident.status = incident_in.status
    incident.tags = tags
    incident.terms = terms
    incident.visibility = incident_in.visibility

    db_session.commit()

    return incident