Feature/vrf

gso/workflows/vrf/modify_vrf_router_list.py

+"""Modify :term:`VRF` to add or remove routers."""
+
+from typing import Annotated
+
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, UUIDstr
+from orchestrator.workflow import StepList, begin, done, step, workflow
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import BaseModel, ConfigDict, Field
+
+from gso.products.product_types.router import Router
+from gso.products.product_types.vrf import VRF
+from gso.utils.helpers import active_router_selector
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    """Modify :term:`VRF` to add or remove routers."""
+    subscription = VRF.from_subscription(subscription_id)
+
+    class RouterSelection(BaseModel):
+        router_id: active_router_selector()  # type: ignore[valid-type]
+
+    class ModifyVRFRouterListForm(FormPage):
+        model_config = ConfigDict(title=f"Modify the {subscription.vrf.vrf_name} VRF to add or remove routers.")
+
+        router_list: Annotated[
+            list[RouterSelection],
+            Field(
+                description="A list of routers to add or remove from the VRF.",
+                min_length=0 if subscription.vrf.vrf_router_list else 1,
+            ),
+        ] = [  # noqa: RUF012
+            RouterSelection(router_id=str(router.owner_subscription_id)) for router in subscription.vrf.vrf_router_list
+        ]
+
+    user_input = yield ModifyVRFRouterListForm
+
+    return user_input.model_dump()
+
+
+@step("Update subscription model")
+def update_subscription_model(subscription: VRF, router_list: list[dict[str, UUIDstr]]) -> State:
+    """Update the database model to update the router list."""
+    subscription.vrf.vrf_router_list = [Router.from_subscription(router["router_id"]).router for router in router_list]
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Modify VRF router list",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def modify_vrf_router_list() -> StepList:
+    """Modify the :term:`VRF` router list."""
+    return begin >> store_process_subscription(Target.MODIFY) >> unsync >> update_subscription_model >> resync >> done