Skip to content

Modify ias

Modification workflow for a IAS subscription.

modify_ias_input_form_generator(subscription_id)

Initial form generator for modifying the custom attributes of an existing IAS subscription.

Source code in gso/workflows/l3_core_service/ias/modify_ias.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def modify_ias_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Initial form generator for modifying the custom attributes of an existing IAS subscription."""
    initial_generator = initial_input_form_generator(subscription_id)
    initial_user_input = yield from initial_generator

    subscription = IAS.from_subscription(subscription_id)

    # Additional IAS step
    class IASExtraForm(FormPage):
        # TODO: remove type hint workaround
        ias_flavor: IASFlavor | str = subscription.ias.ias_flavor

    ias_extra = yield IASExtraForm
    return initial_user_input | ias_extra.model_dump()

modify_ias()

Modify IAS subscription.

Source code in gso/workflows/l3_core_service/ias/modify_ias.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@workflow(
    "Modify IAS",
    initial_input_form=wrap_modify_initial_input_form(modify_ias_input_form_generator),
    target=Target.MODIFY,
)
def modify_ias() -> StepList:
    """Modify IAS subscription."""
    access_port_is_added = conditional(lambda state: state["operation"] == Operation.ADD)
    access_port_is_removed = conditional(lambda state: state["operation"] == Operation.REMOVE)
    access_port_is_modified = conditional(lambda state: state["operation"] == Operation.EDIT)

    return (
        begin
        >> store_process_subscription(Target.MODIFY)
        >> unsync
        >> update_ias_subscription_model
        >> access_port_is_added(create_new_sbp)
        >> access_port_is_removed(remove_old_sbp)
        >> access_port_is_modified(modify_existing_sbp)
        >> resync
        >> done
    )