def update()

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

• 135 lines of code
• 18 McCabe index (conditional complexity)

def update(*, db_session, case: Case, case_in: CaseUpdate, current_user: DispatchUser) -> Case:
    """Updates an existing case."""
    update_data = case_in.dict(
        skip_defaults=True,
        exclude={
            "assignee",
            "case_costs",
            "case_priority",
            "case_severity",
            "case_type",
            "duplicates",
            "incidents",
            "project",
            "related",
            "reporter",
            "status",
            "tags",
            "visibility",
        },
    )

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

    if case_in.case_type:
        if case.case_type.name != case_in.case_type.name:
            case_type = case_type_service.get_by_name(
                db_session=db_session,
                project_id=case.project.id,
                name=case_in.case_type.name,
            )
            if case_type:
                case.case_type = case_type

                event_service.log_case_event(
                    db_session=db_session,
                    source="Dispatch Core App",
                    description=(
                        f"Case type changed to {case_in.case_type.name.lower()} "
                        f"by {current_user.email}"
                    ),
                    dispatch_user_id=current_user.id,
                    case_id=case.id,
                )
            else:
                log.warning(f"Case type with name {case_in.case_type.name.lower()} not found.")

    if case_in.case_severity:
        if case.case_severity.name != case_in.case_severity.name:
            case_severity = case_severity_service.get_by_name(
                db_session=db_session,
                project_id=case.project.id,
                name=case_in.case_severity.name,
            )
            if case_severity:
                case.case_severity = case_severity

                event_service.log_case_event(
                    db_session=db_session,
                    source="Dispatch Core App",
                    description=(
                        f"Case severity changed to {case_in.case_severity.name.lower()} "
                        f"by {current_user.email}"
                    ),
                    dispatch_user_id=current_user.id,
                    case_id=case.id,
                )
            else:
                log.warning(
                    f"Case severity with name {case_in.case_severity.name.lower()} not found."
                )

    if case_in.case_priority:
        if case.case_priority.name != case_in.case_priority.name:
            case_priority = case_priority_service.get_by_name(
                db_session=db_session,
                project_id=case.project.id,
                name=case_in.case_priority.name,
            )
            if case_priority:
                case.case_priority = case_priority

                event_service.log_case_event(
                    db_session=db_session,
                    source="Dispatch Core App",
                    description=(
                        f"Case priority changed to {case_in.case_priority.name.lower()} "
                        f"by {current_user.email}"
                    ),
                    dispatch_user_id=current_user.id,
                    case_id=case.id,
                )
            else:
                log.warning(
                    f"Case priority with name {case_in.case_priority.name.lower()} not found."
                )

    if case.status != case_in.status:
        case.status = case_in.status

        event_service.log_case_event(
            db_session=db_session,
            source="Dispatch Core App",
            description=(
                f"Case status changed to {case_in.status.lower()} by {current_user.email}"
            ),
            dispatch_user_id=current_user.id,
            case_id=case.id,
        )

    if case.visibility != case_in.visibility:
        case.visibility = case_in.visibility

        event_service.log_case_event(
            db_session=db_session,
            source="Dispatch Core App",
            description=(
                f"Case visibility changed to {case_in.visibility.lower()} "
                f"by {current_user.email}"
            ),
            dispatch_user_id=current_user.id,
            case_id=case.id,
        )

    case_costs = []
    for case_cost in case_in.case_costs:
        case_costs.append(
            case_cost_service.get_or_create(db_session=db_session, case_cost_in=case_cost)
        )
    case.case_costs = case_costs

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

    related = []
    for r in case_in.related:
        related.append(get(db_session=db_session, case_id=r.id))
    case.related = related

    duplicates = []
    for d in case_in.duplicates:
        duplicates.append(get(db_session=db_session, case_id=d.id))
    case.duplicates = duplicates

    incidents = []
    for i in case_in.incidents:
        incidents.append(incident_service.get(db_session=db_session, incident_id=i.id))
    case.incidents = incidents

    db_session.commit()

    return case