Skip to content

Create Workflow

A create workflow needs an initial input form generator and defines the steps to create a subscription on a product. The @create_workflow decorator adds some additional steps to the workflow that are always part of a create workflow. The steps of a create workflow in general follow the same pattern, as described below using the create node workflow as an example.

@create_workflow("Create node", initial_input_form=initial_input_form_generator)
def create_node() -> StepList:
    return (
        >> construct_node_model
        >> store_process_subscription(Target.CREATE)
        >> create_node_in_ims
        >> reserve_loopback_addresses
        >> provision_node_in_nrm
  1. Collect input from user (initial_input_form)
  2. Instantiate subscription (construct_node_model):
    1. Create inactive subscription model
    2. assign user input to subscription
    3. transition to subscription to provisioning
  3. Register create process for this subscription (store_process_subscription)
  4. Interact with OSS and/or BSS, in this example
    1. Administer subscription in IMS (create_node_in ims)
    2. Reserve IP addresses in IPAM (reserve_loopback_addresses)
    3. Provision subscription in the network (provision_node_in_nrm)
  5. Transition subscription to active and ‘in sync’ (@create_workflow)

As long as every step remains as idempotent as possible, the work can be divided over fewer or more steps as desired.

Input Form

The input form is created by subclassing the FormPage and add the input fields together with the type and indication if they are optional or not. Additional form settings can be changed via the Config class, like for example the title of the form page.

class CreateNodeForm(FormPage):
    model_config = ConfigDict(title=product_name)

    role_id: NodeRoleChoice
    node_name: str
    node_description: str | None = None

By default, Pydantic validates the input against the specified type and will signal incorrect input and/or missing but required input fields. Type annotations can be used to describe additional constraints, for example a check on the validity of the entered VLAN ID can be specified as shown below, the type Vlan can then be used instead of int.

Vlan = Annotated[int, Ge(2), Le(4094), doc("Allowed VLAN ID range.")]

The node role is defined as type Choice and will be rendered as a dropdown that is filled with a mapping between the role IDs and names as defined in Netbox.

def node_role_selector() -> Choice:
    roles = {str( for role in netbox.get_device_roles()}
    return Choice("RolesEnum", zip(roles.keys(), roles.items()))

NodeRoleChoice: TypeAlias = cast(type[Choice], node_role_selector())

When more than one item needs to be selected, a choice_list can be used to specify the constraints, for example to select two ports for a point-to-point service:

def ports_selector(number_of_ports: int) -> type[list[Choice]]:
    subscriptions = subscriptions_by_product_type("Port", [SubscriptionLifecycle.ACTIVE])
    ports = {str(subscription.subscription_id): subscription.description for subscription in subscriptions)}
    return choice_list(
        Choice("PortsEnum", zip(ports.keys(), ports.items())),

PortsChoiceList: TypeAlias = cast(type[Choice], ports_selector(2))

Extra Validation between dependant fields

Validations between multiple fields is also possible by making use of the Pydantic @model_validator decorator that gives access to all fields. To check if the A and B side of a point-to-point service are not on the same network node one could use:

def separate_nodes(self) -> "SelectNodes":
    if self.node_subscription_id_b == self.node_subscription_id_a:
        raise ValueError("node B cannot be the same as node A")
    return self

For more information on validation, see the Pydantic Validators documentation

Finally, a summary form is shown with the user supplied values. When a value appears to be incorrect, the user can go back to the previous form to correct the mistake, otherwise, when the form is submitted, the workflow is kicked off.

summary_fields = ["role_id", "node_name", "node_description"]
yield from create_summary_form(user_input_dict, product_name, summary_fields)