def update()

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

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

def update(*, db_session, cost_model_in: CostModelUpdate) -> CostModel:
    """Updates a cost model."""
    if not has_unique_plugin_event(cost_model_in):
        raise KeyError("Unable to update cost model. Duplicate plugin event ids detected.")

    cost_model = get_cost_model_by_id(db_session=db_session, cost_model_id=cost_model_in.id)
    if not cost_model:
        raise ValueError("Unable to update cost model. No cost model found with that id.")

    cost_model.name = cost_model_in.name
    cost_model.description = cost_model_in.description
    cost_model.enabled = cost_model_in.enabled
    cost_model.created_at = cost_model_in.created_at
    cost_model.updated_at = (
        cost_model_in.updated_at if cost_model_in.updated_at else datetime.utcnow()
    )

    # Update all recognized activities. Delete all removed activities.
    update_activities = []
    delete_activities = []

    for activity in cost_model.activities:
        updated = False
        for idx_in, activity_in in enumerate(cost_model_in.activities):
            if activity.plugin_event.id == activity_in.plugin_event.id:
                update_activities.append((activity, activity_in))
                cost_model_in.activities.pop(idx_in)
                updated = True
                break
        if updated:
            continue

        # Delete activities that have been removed from the cost model.
        delete_activities.append(activity)

    for activity, activity_in in update_activities:
        activity.response_time_seconds = activity_in.response_time_seconds
        activity.enabled = activity_in.enabled
        activity.plugin_event = plugin_service.get_plugin_event_by_id(
            db_session=db_session, plugin_event_id=activity_in.plugin_event.id
        )

    for activity in delete_activities:
        cost_model_service.delete_cost_model_activity(
            db_session=db_session, cost_model_activity_id=activity.id
        )

    # Create new activities.
    for activity_in in cost_model_in.activities:
        activity_out = cost_model_service.create_cost_model_activity(
            db_session=db_session, cost_model_activity_in=activity_in
        )

        if not activity_out:
            log.error("Failed to create cost model activity. Continuing.")
            continue

        cost_model.activities.append(activity_out)

    db_session.commit()
    return cost_model