diff --git a/gso/workflows/site/base_site_model.py b/gso/workflows/site/base_site_model.py new file mode 100644 index 0000000000000000000000000000000000000000..34061ef8c8634865bec1447a05e6460070485ca9 --- /dev/null +++ b/gso/workflows/site/base_site_model.py @@ -0,0 +1,42 @@ +from pydantic import BaseModel, validator +from pydantic.fields import ModelField +from gso.products.product_blocks.site import SiteTier +from gso.utils.helpers import ( + validate_site_fields_is_unique, + validate_ipv4_or_ipv6, + validate_country_code, + validate_site_name, +) + + +class BaseSiteModel(BaseModel): + site_bgp_community_id: int + site_internal_id: int + site_tier: SiteTier + site_ts_address: str + + @validator("site_ts_address", check_fields=False, allow_reuse=True) + def validate_ts_address(cls, site_ts_address: str) -> str: + """Validate that a terminal server address is valid.""" + validate_ipv4_or_ipv6(site_ts_address) + return site_ts_address + + @validator("site_country_code", check_fields=False, allow_reuse=True) + def country_code_must_exist(cls, country_code: str) -> str: + """Validate that the country code exists.""" + validate_country_code(country_code) + return country_code + + @validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name", check_fields=False, allow_reuse=True) + def validate_unique_fields(cls, value: str, field: ModelField) -> str | int: + """Validate that the internal and :term:`BGP` community IDs are unique.""" + return validate_site_fields_is_unique(field.name, value) + + @validator("site_name", check_fields=False, allow_reuse=True) + def site_name_must_be_valid(cls, site_name: str) -> str: + """Validate the site name. + + The site name must consist of three uppercase letters, followed by an optional single digit. + """ + validate_site_name(site_name) + return site_name \ No newline at end of file diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py index d1b37b91bd2f7b49c6fc48d7073ba52e3e8646c2..e51849531e3109fa3bd1b1e25ab6bf59ae4e3e0b 100644 --- a/gso/workflows/site/create_site.py +++ b/gso/workflows/site/create_site.py @@ -6,25 +6,18 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID from orchestrator.workflow import StepList, done, init, step, workflow from orchestrator.workflows.steps import resync, set_status, store_process_subscription from orchestrator.workflows.utils import wrap_create_initial_input_form -from pydantic import validator -from pydantic.fields import ModelField from gso.products.product_blocks import site as site_pb from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate from gso.products.product_types import site from gso.services.crm import customer_selector -from gso.utils.helpers import ( - validate_country_code, - validate_ipv4_or_ipv6, - validate_site_fields_is_unique, - validate_site_name, -) +from gso.workflows.site.base_site_model import BaseSiteModel def initial_input_form_generator(product_name: str) -> FormGenerator: """Get input from the operator about the new site subscription.""" - class CreateSiteForm(FormPage): + class CreateSiteForm(FormPage, BaseSiteModel): class Config: title = product_name @@ -40,32 +33,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: site_tier: site_pb.SiteTier site_ts_address: str - @validator("site_ts_address", allow_reuse=True) - def validate_ts_address(cls, site_ts_address: str) -> str: - """Validate that a terminal server address is valid.""" - validate_ipv4_or_ipv6(site_ts_address) - return site_ts_address - - @validator("site_country_code", allow_reuse=True) - def country_code_must_exist(cls, country_code: str) -> str: - """Validate that the country code exists.""" - validate_country_code(country_code) - return country_code - - @validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name", allow_reuse=True) - def validate_unique_fields(cls, value: str, field: ModelField) -> str | int: - """Validate that the internal and :term:`BGP` community IDs are unique.""" - return validate_site_fields_is_unique(field.name, value) - - @validator("site_name", allow_reuse=True) - def site_name_must_be_valid(cls, site_name: str) -> str: - """Validate the site name. - - The site name must consist of three uppercase letters, followed by an optional single digit. - """ - validate_site_name(site_name) - return site_name - user_input = yield CreateSiteForm return user_input.dict()