diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py index 1daa2d81a1f0c66debfca6bc4fc35fefa33192bb..7e12d8f3c6d409dff46197d6e77d4805a8608976 100644 --- a/gso/utils/helpers.py +++ b/gso/utils/helpers.py @@ -8,6 +8,7 @@ from uuid import UUID import pycountry from orchestrator.types import UUIDstr from pydantic import BaseModel, field_validator +from pydantic_core.core_schema import ValidationInfo from pydantic_forms.validators import Choice from gso import settings @@ -225,29 +226,15 @@ class BaseSiteValidatorModel(BaseModel): validate_country_code(country_code) return country_code - @field_validator("site_ts_address") - def site_ts_address_must_be_unique(cls, site_ts_address: str) -> str: - """Validate that the internal and :term:`BGP` community IDs are unique.""" - validate_site_fields_is_unique("site_ts_address", site_ts_address) - return site_ts_address - - @field_validator("site_internal_id") - def site_internal_id_must_be_unique(cls, site_internal_id: int) -> int: - """Validate that the internal and :term:`BGP` community IDs are unique.""" - validate_site_fields_is_unique("site_internal_id", site_internal_id) - return site_internal_id + @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name") + def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int: + """Validate that a field is unique.""" + if not info.field_name: + raise ValueError("Field name is required") - @field_validator("site_bgp_community_id") - def site_bgp_community_id_must_be_unique(cls, site_bgp_community_id: int) -> int: - """Validate that the internal and :term:`BGP` community IDs are unique.""" - validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id) - return site_bgp_community_id + validate_site_fields_is_unique(info.field_name, value) - @field_validator("site_name") - def site_name_must_be_unique(cls, site_name: str) -> str: - """Validate that the internal and :term:`BGP` community IDs are unique.""" - validate_site_fields_is_unique("site_name", site_name) - return site_name + return value @field_validator("site_name") def site_name_must_be_valid(cls, site_name: str) -> str: diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py index 93ac6f5156344d9c47db476197dbe66b252d972a..ee48784e9c47757e4282c6e39353f2252edf32bb 100644 --- a/gso/workflows/site/modify_site.py +++ b/gso/workflows/site/modify_site.py @@ -12,6 +12,7 @@ from orchestrator.workflows.steps import ( ) from orchestrator.workflows.utils import wrap_modify_initial_input_form from pydantic import ConfigDict, field_validator +from pydantic_core.core_schema import ValidationInfo from pydantic_forms.validators import ReadOnlyField from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier @@ -37,30 +38,26 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: site_tier: ReadOnlyField(subscription.site.site_tier, default_type=SiteTier) # type: ignore[valid-type] site_ts_address: str | None = subscription.site.site_ts_address + @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id") + def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int: + if not info.field_name: + msg = "Field name must be provided." + raise ValueError(msg) + + if value and value == getattr(subscription.site, info.field_name): + return value + + validate_site_fields_is_unique(info.field_name, value) + + return value + @field_validator("site_ts_address") def validate_ts_address(cls, site_ts_address: str) -> str: if site_ts_address and site_ts_address != subscription.site.site_ts_address: - validate_site_fields_is_unique("site_ts_address", site_ts_address) validate_ipv4_or_ipv6(site_ts_address) return site_ts_address - @field_validator("site_internal_id") - def validate_site_internal_id(cls, site_internal_id: int) -> int: - if site_internal_id == subscription.site.site_internal_id: - return site_internal_id - - validate_site_fields_is_unique("site_internal_id", site_internal_id) - return site_internal_id - - @field_validator("site_bgp_community_id") - def validate_site_bgp_community_id(cls, site_bgp_community_id: int) -> int: - if site_bgp_community_id == subscription.site.site_bgp_community_id: - return site_bgp_community_id - - validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id) - return site_bgp_community_id - user_input = yield ModifySiteForm return user_input.dict()