diff --git a/docs/source/module/utils/types.rst b/docs/source/module/utils/types.rst
new file mode 100644
index 0000000000000000000000000000000000000000..c70c8dd0c61a4fa29cc2f123ec4d0643ab7bdf5d
--- /dev/null
+++ b/docs/source/module/utils/types.rst
@@ -0,0 +1,6 @@
+``gso.utils.types``
+===================
+
+.. automodule:: gso.utils.types
+ :members:
+ :show-inheritance:
diff --git a/gso/utils/types.py b/gso/utils/types.py
new file mode 100644
index 0000000000000000000000000000000000000000..3e1b4091b127d9a572c12c7fad462dc4887de9f7
--- /dev/null
+++ b/gso/utils/types.py
@@ -0,0 +1,9 @@
+"""Define custom types for use across the application."""
+
+from typing import Annotated
+
+from pydantic import AfterValidator
+
+from gso.utils.helpers import validate_tt_number
+
+TTNumber = Annotated[str, AfterValidator(validate_tt_number)]
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 373dfc03b63e378b2889c62aca3496011121f909..90a966efacf6d71a4a0e849c329a40b4c8ad5ee2 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -40,9 +40,9 @@ from gso.utils.helpers import (
validate_interface_name_list,
validate_iptrunk_unique_interface,
validate_router_in_netbox,
- validate_tt_number,
)
from gso.utils.shared_enums import Vendor
+from gso.utils.types import TTNumber
from gso.utils.workflow_steps import prompt_sharepoint_checklist_url
@@ -58,7 +58,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class CreateIptrunkForm(FormPage):
model_config = ConfigDict(title=product_name)
- tt_number: str
+ tt_number: TTNumber
partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
geant_s_sid: str | None = None
iptrunk_description: str | None = None
@@ -66,10 +66,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
iptrunk_speed: PhysicalPortCapacity
iptrunk_number_of_members: int
- @field_validator("tt_number")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
-
initial_user_input = yield CreateIptrunkForm
recommended_minimum_links = calculate_recommended_minimum_links(
initial_user_input.iptrunk_number_of_members, initial_user_input.iptrunk_speed
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index 92e37fd5777105be180674296194ea24ae70cf8f..a45b5eca61144577c5dbf58a251b0b1ca7c76f6d 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -10,11 +10,10 @@ from orchestrator.utils.json import json_dumps
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 field_validator
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.utils.helpers import validate_tt_number
+from gso.utils.types import TTNumber
def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -26,11 +25,7 @@ def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
f"{trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn} to "
f"{trunk.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}"
)
- tt_number: str
-
- @field_validator("tt_number")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
+ tt_number: TTNumber
user_input = yield DeployTWAMPForm
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 908fcdbc1879b64e284914d6347c2ffa05640364..be15848ae73f933fa262211e765b0901313dac66 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -38,9 +38,9 @@ from gso.utils.helpers import (
available_lags_choices,
get_router_vendor,
validate_interface_name_list,
- validate_tt_number,
)
from gso.utils.shared_enums import Vendor
+from gso.utils.types import TTNumber
from gso.utils.workflow_steps import set_isis_to_max
@@ -65,16 +65,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class IPTrunkMigrateForm(FormPage):
model_config = ConfigDict(title=form_title)
- tt_number: str
+ tt_number: TTNumber
replace_side: replaced_side_enum # type: ignore[valid-type]
warning_label: Label = "Are we moving to a different Site?"
migrate_to_different_site: bool = False
restore_isis_metric: bool = True
- @field_validator("tt_number", mode="before")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
-
migrate_form_input = yield IPTrunkMigrateForm
current_routers = [
diff --git a/gso/workflows/iptrunk/modify_isis_metric.py b/gso/workflows/iptrunk/modify_isis_metric.py
index 2e4a4586430cf3698d900924737c352463ed3343..285907b45508249794bb8c5fd486b62ed0b4dac6 100644
--- a/gso/workflows/iptrunk/modify_isis_metric.py
+++ b/gso/workflows/iptrunk/modify_isis_metric.py
@@ -12,6 +12,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
+from gso.utils.types import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -19,7 +20,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription = Iptrunk.from_subscription(subscription_id)
class ModifyIptrunkForm(FormPage):
- tt_number: str
+ tt_number: TTNumber
isis_metric: int = subscription.iptrunk.iptrunk_isis_metric
user_input = yield ModifyIptrunkForm
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index f5d17d7752503d8fdb8979d0eefadd7818af4936..394e369a88d1f64481750080410ffcb8f4335064 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -32,9 +32,9 @@ from gso.utils.helpers import (
get_router_vendor,
validate_interface_name_list,
validate_iptrunk_unique_interface,
- validate_tt_number,
)
from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, Vendor
+from gso.utils.types import TTNumber
from gso.workflows.iptrunk.migrate_iptrunk import check_ip_trunk_optical_levels_pre
from gso.workflows.iptrunk.validate_iptrunk import check_ip_trunk_isis
@@ -85,7 +85,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription = Iptrunk.from_subscription(subscription_id)
class ModifyIptrunkForm(FormPage):
- tt_number: str
+ tt_number: TTNumber
geant_s_sid: str | None = subscription.iptrunk.geant_s_sid
iptrunk_description: str | None = subscription.iptrunk.iptrunk_description
iptrunk_type: IptrunkType = subscription.iptrunk.iptrunk_type
@@ -103,10 +103,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
str(subscription.iptrunk.iptrunk_ipv6_network), default_type=IPv6AddressType
)
- @field_validator("tt_number")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
-
initial_user_input = yield ModifyIptrunkForm
recommended_minimum_links = calculate_recommended_minimum_links(
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index bee9739a9732ea3db1415e8d7955886b9dcaa2bb..bb1a6fd90b3d9a5e3b9aa9bb633db58cc2eb1cd4 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -16,15 +16,15 @@ from orchestrator.workflows.steps import (
unsync,
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import field_validator
from gso.products.product_blocks.iptrunk import IptrunkSideBlock
from gso.products.product_types.iptrunk import Iptrunk
from gso.services import infoblox
from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
-from gso.utils.helpers import get_router_vendor, validate_tt_number
+from gso.utils.helpers import get_router_vendor
from gso.utils.shared_enums import Vendor
+from gso.utils.types import TTNumber
from gso.utils.workflow_steps import set_isis_to_max
@@ -40,16 +40,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
- tt_number: str
+ tt_number: TTNumber
termination_label: Label = (
"Please confirm whether configuration should get removed from the A and B sides of the trunk."
)
remove_configuration: bool = True
- @field_validator("tt_number")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
-
user_input = yield TerminateForm
return user_input.model_dump()
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 8382e227181233c145371f00aeb835ba91797f58..59dd6210578d9556fdf79ecacf7df0b5518fdbd5 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -25,6 +25,7 @@ from gso.services.sharepoint import SharePointClient
from gso.settings import load_oss_params
from gso.utils.helpers import generate_fqdn, iso_from_ipv4
from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.types import TTNumber
from gso.utils.workflow_steps import (
deploy_base_config_dry,
deploy_base_config_real,
@@ -48,7 +49,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
class CreateRouterForm(FormPage):
model_config = ConfigDict(title=product_name)
- tt_number: str
+ tt_number: TTNumber
partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
vendor: Vendor
router_site: _site_selector() # type: ignore[valid-type]
diff --git a/gso/workflows/router/promote_p_to_pe.py b/gso/workflows/router/promote_p_to_pe.py
index 5b3a4bdbc2cfa0a8d7458298f1464a56993278d7..c8c4f016b185fc544e8b013011dbb88e66fdf1d9 100644
--- a/gso/workflows/router/promote_p_to_pe.py
+++ b/gso/workflows/router/promote_p_to_pe.py
@@ -11,15 +11,16 @@ from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, begin, conditional, done, inputstep, 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 ConfigDict, field_validator
+from pydantic import ConfigDict
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import Router
from gso.services import lso_client
from gso.services.kentik_client import KentikClient, NewKentikDevice
from gso.services.lso_client import lso_interaction
-from gso.utils.helpers import generate_inventory_for_active_routers, validate_tt_number
+from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
+from gso.utils.types import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -29,23 +30,11 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class PromotePToPEForm(FormPage):
model_config = ConfigDict(title=f"Promote {subscription.router.router_fqdn} to PE router?")
- tt_number: str
-
- @field_validator("tt_number")
- def validate_tt_number(cls, tt_number: str) -> str:
- return validate_tt_number(tt_number)
+ tt_number: TTNumber
user_input = yield PromotePToPEForm
- return user_input.model_dump()
-
-
-@step("Prepare required keys in state")
-def prepare_state(subscription_id: UUIDstr) -> State:
- """Add required keys to the state for the workflow to run successfully."""
- router = Router.from_subscription(subscription_id)
-
- return {"subscription": router}
+ return user_input.model_dump() | {"subscription": subscription}
@step("Evacuate the router by setting isis_overload")
@@ -138,8 +127,8 @@ def create_kentik_device(subscription: Router) -> State:
kentik_site = kentik_client.get_site_by_name(subscription.router.router_site.site_name)
if not kentik_site:
- msg = f"Site could not be found in Kentik: {subscription.router.router_site.site_name}"
- raise ProcessFailureError(msg)
+ msg = "Site could not be found in Kentik."
+ raise ProcessFailureError(msg, details=subscription.router.router_site.site_name)
site_tier = subscription.router.router_site.site_tier
new_device = NewKentikDevice(
@@ -573,7 +562,6 @@ def promote_p_to_pe() -> StepList:
return (
begin
>> store_process_subscription(Target.MODIFY)
- >> prepare_state
>> router_is_juniper(done)
>> router_is_pe(done)
>> unsync
diff --git a/gso/workflows/router/redeploy_base_config.py b/gso/workflows/router/redeploy_base_config.py
index c1a24c8340dc6c129fa99c1bb93be528bd85bd18..b30d02f16b6bb197a20257f9016eabb6dad0d8fa 100644
--- a/gso/workflows/router/redeploy_base_config.py
+++ b/gso/workflows/router/redeploy_base_config.py
@@ -10,6 +10,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.router import Router
from gso.services.lso_client import lso_interaction
+from gso.utils.types import TTNumber
from gso.utils.workflow_steps import deploy_base_config_dry, deploy_base_config_real
@@ -18,7 +19,7 @@ def _initial_input_form(subscription_id: UUIDstr) -> FormGenerator:
class RedeployBaseConfigForm(FormPage):
info_label: Label = f"Redeploy base config on {router.router.router_fqdn}?"
- tt_number: str
+ tt_number: TTNumber
user_input = yield RedeployBaseConfigForm
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index 688e3ecc21543ddc5ca678296b763fe7c6194bb5..07b0692589262b0b35e80c13a6d42dc51cdfec07 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -26,6 +26,7 @@ from gso.services.lso_client import execute_playbook, lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.utils.helpers import generate_inventory_for_active_routers
from gso.utils.shared_enums import Vendor
+from gso.utils.types import TTNumber
logger = logging.getLogger(__name__)
@@ -42,7 +43,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
- tt_number: str
+ tt_number: TTNumber
termination_label: Label = "Please confirm whether configuration should get removed from the router."
remove_configuration: bool = True
update_ibgp_mesh_label: Label = "Please confirm whether the iBGP mesh should get updated."
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index 58d207029c9431d7252ddf2b5c87e171282944ba..8fbb2813c94cff8443647a033c146e93f34b7ca4 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -18,6 +18,7 @@ from gso.services import librenms_client, lso_client
from gso.services.lso_client import lso_interaction
from gso.services.subscriptions import get_trunks_that_terminate_on_router
from gso.utils.helpers import SNMPVersion, generate_inventory_for_active_routers
+from gso.utils.types import TTNumber
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -31,7 +32,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class AddBGPSessionForm(FormPage):
model_config = ConfigDict(title=f"Add {subscription.router.router_fqdn} to the iBGP mesh?")
- tt_number: str
+ tt_number: TTNumber
@model_validator(mode="before")
def router_has_a_trunk(cls, data: Any) -> Any: