diff --git a/gso/api/v1/imports.py b/gso/api/v1/imports.py
index 0b2b6b1624c06acb657cca57ebb03a0817972cf0..bde96f2253572c1350cc0a550f33debe19d799d2 100644
--- a/gso/api/v1/imports.py
+++ b/gso/api/v1/imports.py
@@ -1,13 +1,13 @@
""":term:`GSO` :term:`API` endpoints that import different types of existing services."""
import ipaddress
-from typing import Any
+from typing import Any, Self
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.routing import APIRouter
from orchestrator.services import processes
-from pydantic import BaseModel, root_validator, validator
+from pydantic import BaseModel, field_validator, model_validator
from gso.auth.security import opa_security_default
from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
@@ -87,7 +87,7 @@ class IptrunkImportModel(BaseModel):
for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id"])
}
- @validator("partner")
+ @field_validator("partner")
def check_if_partner_exists(cls, value: str) -> str:
"""Validate that the partner exists."""
try:
@@ -98,7 +98,7 @@ class IptrunkImportModel(BaseModel):
return value
- @validator("side_a_node_id", "side_b_node_id")
+ @field_validator("side_a_node_id", "side_b_node_id")
def check_if_router_side_is_available(cls, value: str) -> str:
"""Both sides of the trunk must exist in :term:`GSO`."""
if value not in cls._get_active_routers():
@@ -107,7 +107,7 @@ class IptrunkImportModel(BaseModel):
return value
- @validator("side_a_ae_members", "side_b_ae_members")
+ @field_validator("side_a_ae_members", "side_b_ae_members")
def check_side_uniqueness(cls, value: list[str]) -> list[str]:
""":term:`LAG` members must be unique."""
if len(value) != len(set(value)):
@@ -116,25 +116,21 @@ class IptrunkImportModel(BaseModel):
return value
- @root_validator
- def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
+ @model_validator(mode="after")
+ def check_members(self) -> Self:
"""Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""
- min_links = values["iptrunk_minimum_links"]
- side_a_members = values.get("side_a_ae_members", [])
- side_b_members = values.get("side_b_ae_members", [])
+ len_a = len(self.side_a_ae_members)
+ len_b = len(self.side_b_ae_members)
- len_a = len(side_a_members)
- len_b = len(side_b_members)
-
- if len_a < min_links:
- msg = f"Side A members should be at least {min_links} (iptrunk_minimum_links)"
+ if len_a < self.iptrunk_minimum_links:
+ msg = f"Side A members should be at least {self.iptrunk_minimum_links} (iptrunk_minimum_links)"
raise ValueError(msg)
if len_a != len_b:
msg = "Mismatch between Side A and B members"
raise ValueError(msg)
- return values
+ return self
class SuperPopSwitchImportModel(BaseModel):
diff --git a/gso/auth/opa.py b/gso/auth/opa.py
new file mode 100644
index 0000000000000000000000000000000000000000..28c0cad8feff8d74207cda86078fdea0318b46ff
--- /dev/null
+++ b/gso/auth/opa.py
@@ -0,0 +1,44 @@
+from http import HTTPStatus
+
+from fastapi.exceptions import HTTPException
+from fastapi.params import Depends
+from httpx import AsyncClient, NetworkError
+from oauth2_lib.fastapi import OIDCUserModel, OPAAuthorization, OPAResult
+from oauth2_lib.settings import oauth2lib_settings
+from starlette.requests import Request
+from structlog import get_logger
+
+from gso.auth.oidc import oidc_instance
+
+logger = get_logger(__name__)
+
+
+class OPAAuthorization(OPAAuthorization):
+ _instance = None
+
+ def __new__(cls, *args, **kwargs):
+ if cls._instance is None:
+ cls._instance = super(OPAAuthorization, cls).__new__(cls)
+ return cls._instance
+
+ async def authorize(
+ self, request: Request, user_info: OIDCUserModel = Depends(oidc_instance.authenticate)
+ ) -> bool | None:
+ return await super().authorize(request, user_info)
+
+ async def get_decision(self, async_request: AsyncClient, opa_input: dict) -> OPAResult:
+ logger.debug("Posting input json to Policy agent", opa_url=self.opa_url, input=opa_input)
+ try:
+ result = await async_request.post(self.opa_url, json=opa_input)
+ except (NetworkError, TypeError) as exc:
+ logger.debug("Could not get decision from policy agent", error=str(exc))
+ raise HTTPException(status_code=HTTPStatus.SERVICE_UNAVAILABLE, detail="Policy agent is unavailable")
+
+ json_result = result.json()
+ logger.debug("Received decision from policy agent", decision=json_result)
+ return OPAResult(decision_id=json_result["decision_id"], result=json_result["result"]["allow"])
+
+
+opa_instance = OPAAuthorization(
+ opa_url=oauth2lib_settings.OPA_URL,
+)
diff --git a/gso/auth/settings.py b/gso/auth/settings.py
index 29c1fc806a8589b38158a3f95dddf3f10cb8bdf3..b3ab1a6a569e2e594e181c23c231366e212f4905 100644
--- a/gso/auth/settings.py
+++ b/gso/auth/settings.py
@@ -6,7 +6,8 @@ with external authentication providers for enhanced security management.
Todo: Remove token and sensitive data from OPA console and API.
"""
-from pydantic import BaseSettings, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings
class Oauth2LibSettings(BaseSettings):
diff --git a/gso/migrations/env.py b/gso/migrations/env.py
index 45dc109d4786205b3359743edf3681283ca58797..968abeb94a1145de0c923cdc8d27dd2030a55df7 100644
--- a/gso/migrations/env.py
+++ b/gso/migrations/env.py
@@ -15,7 +15,7 @@ config = context.config
# This line sets up loggers basically.
logger = logging.getLogger("alembic.env")
-config.set_main_option("sqlalchemy.url", app_settings.DATABASE_URI)
+config.set_main_option("sqlalchemy.url", str(app_settings.DATABASE_URI))
target_metadata = BaseModel.metadata
diff --git a/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py b/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py
new file mode 100644
index 0000000000000000000000000000000000000000..aa9593a8ba0279329a6361900a1965b2eddc365c
--- /dev/null
+++ b/gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py
@@ -0,0 +1,23 @@
+"""remove subscription cancellation workflow.
+
+Revision ID: 1ec810b289c0
+Revises:
+Create Date: 2024-04-02 10:21:08.539591
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '1ec810b289c0'
+down_revision = '4ec89ab289c0'
+branch_labels = None
+# TODO: check it carefuly
+depends_on = '048219045729' # in this revision, SURF has added a new columns to the workflow table like delted_at, so we need to add a dependency on the revision that added the columns to the workflow table.
+
+
+def upgrade() -> None:
+ pass
+
+
+def downgrade() -> None:
+ pass
+
diff --git a/gso/products/product_blocks/iptrunk.py b/gso/products/product_blocks/iptrunk.py
index 901f37e787805e68307fc598a95080b0e6cdbd08..b9442fad884d889659853acca92a0ebd542fc512 100644
--- a/gso/products/product_blocks/iptrunk.py
+++ b/gso/products/product_blocks/iptrunk.py
@@ -1,11 +1,13 @@
"""IP trunk product block that has all parameters of a subscription throughout its lifecycle."""
import ipaddress
-from typing import TypeVar
+from typing import Annotated, TypeVar
-from orchestrator.domain.base import ProductBlockModel
-from orchestrator.forms.validators import UniqueConstrainedList
+from annotated_types import Len
+from orchestrator.domain.base import ProductBlockModel, T
from orchestrator.types import SubscriptionLifecycle, strEnum
+from pydantic import AfterValidator
+from pydantic_forms.validators import validate_unique_list
from gso.products.product_blocks.router import (
RouterBlock,
@@ -33,11 +35,8 @@ class IptrunkType(strEnum):
LEASED = "Leased"
-T_co = TypeVar("T_co", covariant=True)
-
-
-class LAGMemberList(UniqueConstrainedList[T_co]): # type: ignore[type-var]
- """A list of :term:`LAG` member interfaces."""
+# A list of :term:`LAG` member interfaces.
+LAGMemberList = Annotated[list[T], AfterValidator(validate_unique_list), Len(min_length=0, max_length=None)]
class IptrunkInterfaceBlockInactive(
@@ -65,11 +64,14 @@ class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[Subscr
interface_description: str
-class IptrunkSides(UniqueConstrainedList[T_co]): # type: ignore[type-var]
- """A list of IP trunk interfaces that make up one side of a link."""
+def validate_unique_list(value):
+ if len(value) != len(set(value)):
+ raise ValueError("List items must be unique")
+ return value
- min_items = 2
- max_items = 2
+
+T_co = TypeVar("T_co", covariant=True)
+IptrunkSides = Annotated[list[T_co], AfterValidator(validate_unique_list), Len(min_length=2, max_length=2)]
class IptrunkSideBlockInactive(
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index 1852b24615076b2d76dde41db71a9e5d5fcc535f..8082abaa29f41ba9c3afd4dfae2fee98930cfb68 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -1,10 +1,11 @@
"""The product block that describes a site subscription."""
import re
+from typing import Annotated
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle, strEnum
-from pydantic import ConstrainedStr
+from pydantic import AfterValidator, Field
class SiteTier(strEnum):
@@ -20,44 +21,41 @@ class SiteTier(strEnum):
TIER4 = 4
-class LatitudeCoordinate(ConstrainedStr):
- """A latitude coordinate, modeled as a constrained string.
-
- The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. It can be a
- floating-point number or an integer.
- Valid examples: 40.7128, -74.0060, 90, -90, 0
- """
-
+def validate_latitude(v: float) -> float:
regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
-
- @classmethod
- def validate(cls, value: str) -> str:
- """Validate that a latitude coordinate is valid."""
- if not cls.regex.match(value):
- msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
- raise ValueError(msg)
-
- return value
+ if not regex.match(str(v)):
+ raise ValueError("Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'.")
+ return v
-class LongitudeCoordinate(ConstrainedStr):
- """A longitude coordinate, modeled as a constrained string.
-
- The coordinate must match the format conforming to the longitude
- range of -180 to +180 degrees. It can be a floating point number or an integer.
- Valid examples: 40.7128, -74.0060, 180, -180, 0
- """
-
+def validate_longitude(v: float) -> float:
regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
-
- @classmethod
- def validate(cls, value: str) -> str:
- """Validate that a longitude coordinate is valid."""
- if not cls.regex.match(value):
- msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180'"
- raise ValueError(msg)
-
- return value
+ if not regex.match(str(v)):
+ raise ValueError("Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'.")
+
+ return v
+
+
+LatitudeCoordinate = Annotated[
+ float,
+ Field(
+ ge=-90,
+ le=90,
+ example="40.7128",
+ description="A latitude coordinate, modeled as a string. The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0.",
+ ),
+ AfterValidator(validate_latitude)
+]
+LongitudeCoordinate = Annotated[
+ float,
+ Field(
+ ge=-180,
+ le=180,
+ example="74.0060",
+ description="A longitude coordinate, modeled as a string. The coordinate must match the format conforming to the longitude range of -180 to +180 degrees. It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 180, -180, 0.",
+ ),
+ AfterValidator(validate_longitude)
+]
class SiteBlockInactive(
diff --git a/gso/schema/partner.py b/gso/schema/partner.py
index 890adcb9b20b08f6c244e8986ad20eaf4def83fc..d8aa1b741a38d552ec625bfaa9c81905c2e32ab0 100644
--- a/gso/schema/partner.py
+++ b/gso/schema/partner.py
@@ -3,7 +3,7 @@
from datetime import datetime
from uuid import uuid4
-from pydantic import BaseModel, EmailStr, Field
+from pydantic import BaseModel, ConfigDict, EmailStr, Field
from gso.db.models import PartnerType
@@ -23,8 +23,4 @@ class PartnerCreate(BaseModel):
partner_type: PartnerType
created_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
updated_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
-
- class Config:
- """Pydantic model configuration."""
-
- orm_mode = True
+ model_config = ConfigDict(from_attributes=True)
diff --git a/gso/services/lso_client.py b/gso/services/lso_client.py
index 67e4b77cce1e4c03f1ba200b6633079cb2476437..9f7024649b9c4d7a394d18e791003f6a2beaa828 100644
--- a/gso/services/lso_client.py
+++ b/gso/services/lso_client.py
@@ -13,9 +13,10 @@ from orchestrator.config.assignee import Assignee
from orchestrator.types import State
from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import Step, StepList, begin, callback_step, inputstep
-from pydantic_forms.core import FormPage, ReadOnlyField
+from pydantic import ConfigDict
+from pydantic_forms.core import FormPage
from pydantic_forms.types import FormGenerator
-from pydantic_forms.validators import Label, LongText
+from pydantic_forms.validators import Label, LongText, ReadOnlyField
from gso import settings
@@ -125,8 +126,7 @@ def _show_results(state: State) -> FormGenerator:
return state
class ConfirmRunPage(FormPage):
- class Config:
- title: str = state["lso_result_title"]
+ model_config = ConfigDict()
if "lso_result_extra_label" in state:
extra_label: Label = state["lso_result_extra_label"]
diff --git a/gso/settings.py b/gso/settings.py
index ced74ba5c01f59555976396dc43ad268bc786c0b..7f601e94d33fef2ebe3b6e426a0cedd92adb00e9 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -10,7 +10,8 @@ import logging
import os
from pathlib import Path
-from pydantic import BaseSettings, NonNegativeInt
+from pydantic import NonNegativeInt
+from pydantic_settings import BaseSettings
logger = logging.getLogger(__name__)
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 6c30324ed0b81064bdc4c84e862f1a0ff671b9da..0a87cd51ec98bef4a3fff0e63c214f0f19900029 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -8,8 +8,7 @@ from uuid import UUID
import pycountry
from orchestrator.types import UUIDstr
-from pydantic import BaseModel, validator
-from pydantic.fields import ModelField
+from pydantic import BaseModel, field_validator
from pydantic_forms.validators import Choice
from gso import settings
@@ -211,31 +210,46 @@ class BaseSiteValidatorModel(BaseModel):
site_tier: SiteTier
site_ts_address: str
- @validator("site_ts_address", check_fields=False, allow_reuse=True)
+ @classmethod
+ @field_validator("site_ts_address", check_fields=False)
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)
+ @classmethod
+ @field_validator("site_country_code", check_fields=False)
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:
+ @classmethod
+ @field_validator("site_ts_address", check_fields=False)
+ def site_ts_address_must_be_unique(cls, site_ts_address: str) -> str:
+ """Validate that the internal and :term:`BGP` community IDs are unique."""
+ return validate_site_fields_is_unique("site_ts_address", site_ts_address)
+
+ @classmethod
+ @field_validator("site_internal_id", check_fields=False)
+ def site_internal_id_must_be_unique(cls, site_internal_id: int) -> int:
+ """Validate that the internal and :term:`BGP` community IDs are unique."""
+ return validate_site_fields_is_unique("site_internal_id", site_internal_id)
+
+ @classmethod
+ @field_validator("site_bgp_community_id", check_fields=False)
+ 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."""
+ return validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id)
+
+ @classmethod
+ @field_validator("site_name", check_fields=False)
+ def site_name_must_be_unique(cls, site_name: str) -> str:
"""Validate that the internal and :term:`BGP` community IDs are unique."""
- return validate_site_fields_is_unique(field.name, value)
+ return validate_site_fields_is_unique("site_name", site_name)
- @validator("site_name", check_fields=False, allow_reuse=True)
+ @classmethod
+ @field_validator("site_name", check_fields=False)
def site_name_must_be_valid(cls, site_name: str) -> str:
"""Validate the site name.
diff --git a/gso/utils/shared_enums.py b/gso/utils/shared_enums.py
index c0116e1690d6384cabd9ce16cf1ee79201a0d6b8..c5d6fa6d9b54c970830c2804e436f4010f9b3dbb 100644
--- a/gso/utils/shared_enums.py
+++ b/gso/utils/shared_enums.py
@@ -1,6 +1,8 @@
"""Shared choices for the different models."""
-from pydantic import ConstrainedInt
+from typing import Annotated
+
+from pydantic import Field
from pydantic_forms.types import strEnum
@@ -11,14 +13,17 @@ class Vendor(strEnum):
NOKIA = "nokia"
-class PortNumber(ConstrainedInt):
- """Constrained integer for valid port numbers.
-
- The range from 49152 to 65535 is marked as ephemeral, and can therefore not be selected for permanent allocation.
- """
-
- gt = 0
- le = 49151
+PortNumber = Annotated[
+ int,
+ Field(
+ gt=0,
+ le=49151,
+ description=(
+ "Constrained integer for valid port numbers. The range from 49152 to 65535 is marked as ephemeral, "
+ "and can therefore not be selected for permanent allocation."
+ ),
+ ),
+]
class ConnectionStrategy(strEnum):
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 386fb36bf3d1b61e2d0a52252fc832d301a7a05c..dd7b070abb7b3c86263c92a91b38e1d5404ddeaa 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -1,19 +1,21 @@
"""A creation workflow that deploys a new IP trunk service."""
import json
+from typing import Annotated
from uuid import uuid4
+from annotated_types import Len
from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, Label, UniqueConstrainedList
+from orchestrator.forms.validators import Choice, Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, conditional, done, init, inputstep, 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_forms.core import ReadOnlyField
+from pydantic import AfterValidator, ConfigDict, field_validator
+from pydantic_forms.validators import ReadOnlyField, validate_unique_list
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import (
@@ -52,8 +54,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
routers[str(router["subscription_id"])] = router["description"]
class CreateIptrunkForm(FormPage):
- class Config:
- title = product_name
+ model_config = ConfigDict(title=product_name)
tt_number: str
partner: str = ReadOnlyField("GEANT")
@@ -63,7 +64,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
iptrunk_speed: PhysicalPortCapacity
iptrunk_minimum_links: int
- @validator("tt_number", allow_reuse=True)
+ @field_validator("tt_number")
+ @classmethod
def validate_tt_number(cls, tt_number: str) -> str:
return validate_tt_number(tt_number)
@@ -72,20 +74,18 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class SelectRouterSideA(FormPage):
- class Config:
- title = "Select a router for side A of the trunk."
+ model_config = ConfigDict(title="Select a router for side A of the trunk.")
side_a_node_id: router_enum_a # type: ignore[valid-type]
- @validator("side_a_node_id", allow_reuse=True)
+ @field_validator("side_a_node_id")
def validate_device_exists_in_netbox(cls, side_a_node_id: UUIDstr) -> str | None:
return validate_router_in_netbox(side_a_node_id)
user_input_router_side_a = yield SelectRouterSideA
router_a = user_input_router_side_a.side_a_node_id.name
- class JuniperAeMembers(UniqueConstrainedList[LAGMember]):
- min_items = initial_user_input.iptrunk_minimum_links
+ JuniperAeMembers = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
if get_router_vendor(router_a) == Vendor.NOKIA:
@@ -95,22 +95,19 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
initial_user_input.iptrunk_speed,
)
- class NokiaAeMembersA(UniqueConstrainedList[NokiaLAGMemberA]):
- min_items = initial_user_input.iptrunk_minimum_links
-
- ae_members_side_a = NokiaAeMembersA
+ ae_members_side_a = Annotated[list[NokiaLAGMemberA], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
else:
ae_members_side_a = JuniperAeMembers # type: ignore[assignment]
class CreateIptrunkSideAForm(FormPage):
- class Config:
- title = "Provide subscription details for side A of the trunk."
+ model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")
side_a_ae_iface: available_lags_choices(router_a) or str # type: ignore[valid-type]
side_a_ae_geant_a_sid: str
side_a_ae_members: ae_members_side_a # type: ignore[valid-type]
- @validator("side_a_ae_members", allow_reuse=True)
+ @field_validator("side_a_ae_members")
+ @classmethod
def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
validate_iptrunk_unique_interface(side_a_ae_members)
vendor = get_router_vendor(router_a)
@@ -123,12 +120,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
router_enum_b = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class SelectRouterSideB(FormPage):
- class Config:
- title = "Select a router for side B of the trunk."
+ model_config = ConfigDict(title="Select a router for side B of the trunk.")
side_b_node_id: router_enum_b # type: ignore[valid-type]
- @validator("side_b_node_id", allow_reuse=True)
+ @field_validator("side_b_node_id")
+ @classmethod
def validate_device_exists_in_netbox(cls, side_b_node_id: UUIDstr) -> str | None:
return validate_router_in_netbox(side_b_node_id)
@@ -143,24 +140,19 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
initial_user_input.iptrunk_speed,
)
- class NokiaAeMembersB(UniqueConstrainedList):
- min_items = len(user_input_side_a.side_a_ae_members)
- max_items = len(user_input_side_a.side_a_ae_members)
- item_type = NokiaLAGMemberB
-
- ae_members_side_b = NokiaAeMembersB
+ ae_members_side_b = Annotated[list[NokiaLAGMemberB], AfterValidator(validate_unique_list), Len(min_length=len(user_input_side_a.side_a_ae_members), max_length= len(user_input_side_a.side_a_ae_members))]
else:
ae_members_side_b = JuniperAeMembers # type: ignore[assignment]
class CreateIptrunkSideBForm(FormPage):
- class Config:
- title = "Provide subscription details for side B of the trunk."
+ model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")
side_b_ae_iface: available_lags_choices(router_b) or str # type: ignore[valid-type]
side_b_ae_geant_a_sid: str
side_b_ae_members: ae_members_side_b # type: ignore[valid-type]
- @validator("side_b_ae_members", allow_reuse=True)
+ @field_validator("side_b_ae_members")
+ @classmethod
def validate_side_b_ae_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
validate_iptrunk_unique_interface(side_b_ae_members)
vendor = get_router_vendor(router_b)
@@ -479,8 +471,7 @@ def prompt_start_new_checklist(subscription: IptrunkProvisioning) -> FormGenerat
oss_params = load_oss_params()
class SharepointPrompt(FormPage):
- class Config:
- title = "Start new checklist"
+ model_config = ConfigDict(title="Start new checklist")
info_label_1: Label = (
f"Visit {oss_params.SHAREPOINT.checklist_site_url} and start a new Sharepoint checklist for an IPtrunk " # type: ignore[assignment]
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index b9003078a016313bab3012cfee8eaf3f11f278ab..011b37c85f82c486dccb87c3a2d1627ae852be44 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -7,7 +7,7 @@ from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.workflow import StepList, done, init, 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 validator
+from pydantic import field_validator
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.lso_client import execute_playbook, lso_interaction
@@ -25,7 +25,8 @@ def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
tt_number: str
- @validator("tt_number", allow_reuse=True)
+ @field_validator("tt_number")
+ @classmethod
def validate_tt_number(cls, tt_number: str) -> str:
return validate_tt_number(tt_number)
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index add6d5f01d6c5d74f31dc426453856f4240bf963..fe97d66ecde2d019649cea61bf704225cd578b87 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -7,12 +7,14 @@ configured to run from A to C. B is then no longer associated with this IP trunk
import copy
import json
import re
+from typing import Annotated
from uuid import uuid4
+from annotated_types import Len
from orchestrator import step, workflow
from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, Label, UniqueConstrainedList
+from orchestrator.forms.validators import Choice, Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.utils.errors import ProcessFailureError
@@ -20,8 +22,8 @@ from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, conditional, done, init, inputstep
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import validator
-from pydantic_forms.core import ReadOnlyField
+from pydantic import AfterValidator, ConfigDict, field_validator
+from pydantic_forms.validators import ReadOnlyField, validate_unique_list
from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
@@ -63,8 +65,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
class IPTrunkMigrateForm(FormPage):
- class Config:
- title = form_title
+ model_config = ConfigDict(title=form_title)
tt_number: str
replace_side: replaced_side_enum # type: ignore[valid-type]
@@ -72,7 +73,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
migrate_to_different_site: bool = False
restore_isis_metric: bool = True
- @validator("tt_number", allow_reuse=True, pre=True, always=True)
+ @field_validator("tt_number", mode="before")
def validate_tt_number(cls, tt_number: str) -> str:
return validate_tt_number(tt_number)
@@ -100,8 +101,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
new_router_enum = Choice("Select a new router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class NewSideIPTrunkRouterForm(FormPage):
- class Config:
- title = form_title
+ model_config = ConfigDict(title=form_title)
new_node: new_router_enum # type: ignore[valid-type]
@@ -118,18 +118,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription.iptrunk.iptrunk_speed,
)
- class NokiaAeMembers(UniqueConstrainedList[NokiaLAGMember]):
- min_items = len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
- max_items = len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members)
-
- ae_members = NokiaAeMembers
+ ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members), max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members))]
else:
- class JuniperLagMember(UniqueConstrainedList[LAGMember]):
- min_items = len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
- max_items = len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members)
-
- ae_members = JuniperLagMember # type: ignore[assignment]
+ ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members), max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members))]
replace_index = (
0
@@ -146,14 +138,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
]
class NewSideIPTrunkForm(FormPage):
- class Config:
- title = form_title
+ model_config = ConfigDict(title=form_title)
new_lag_interface: side_a_ae_iface # type: ignore[valid-type]
existing_lag_interface: list[LAGMember] = ReadOnlyField(existing_lag_ae_members)
new_lag_member_interfaces: ae_members # type: ignore[valid-type]
- @validator("new_lag_interface", allow_reuse=True, pre=True, always=True)
+ @field_validator("new_lag_interface", mode="before")
def lag_interface_proper_name(cls, new_lag_interface: str) -> str:
if get_router_vendor(new_router) == Vendor.JUNIPER:
juniper_lag_re = re.compile("^ae\\d{1,2}$")
@@ -162,7 +153,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
raise ValueError(msg)
return new_lag_interface
- @validator("new_lag_member_interfaces", allow_reuse=True, pre=True, always=True)
+ @field_validator("new_lag_member_interfaces", mode="before")
def is_interface_names_valid_juniper(cls, new_lag_member_interfaces: list[LAGMember]) -> list[LAGMember]:
vendor = get_router_vendor(new_router)
return validate_interface_name_list(new_lag_member_interfaces, vendor)
@@ -394,8 +385,7 @@ def confirm_continue_move_fiber() -> FormGenerator:
"""Wait for confirmation from an operator that the physical fiber has been moved."""
class ProvisioningResultPage(FormPage):
- class Config:
- title = "Please confirm before continuing"
+ model_config = ConfigDict(title="Please confirm before continuing")
info_label: Label = "New trunk interface has been deployed, wait for the physical connection to be moved." # type: ignore[assignment]
@@ -484,8 +474,7 @@ def confirm_continue_restore_isis() -> FormGenerator:
"""Wait for an operator to confirm that the old :term:`ISIS` metric should be restored."""
class ProvisioningResultPage(FormPage):
- class Config:
- title = "Please confirm before continuing"
+ model_config = ConfigDict(title="Please confirm before continuing")
info_label: Label = "ISIS config has been deployed, confirm if you want to restore the old metric." # type: ignore[assignment]
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index d3b5e60ee539660e7aa2fbd22088d20c8235e6cb..19edbdf60be481b5a1b3a02432920d82cf418c0f 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -2,18 +2,19 @@
import ipaddress
import json
+from typing import Annotated
from uuid import UUID, uuid4
-from orchestrator.forms import FormPage, ReadOnlyField
-from orchestrator.forms.validators import UniqueConstrainedList
+from annotated_types import Len
+from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.utils.json import json_dumps
from orchestrator.workflow import StepList, conditional, done, init, 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 validator
-from pydantic_forms.validators import Label
+from pydantic import AfterValidator, ConfigDict, field_validator
+from pydantic_forms.validators import Label, ReadOnlyField, validate_unique_list
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlock,
@@ -60,16 +61,10 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
)
)
- class NokiaAeMembers(UniqueConstrainedList[NokiaLAGMember]):
- min_items = iptrunk_minimum_link
-
- ae_members = NokiaAeMembers
+ ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
else:
- class JuniperAeMembers(UniqueConstrainedList[LAGMember]):
- min_items = iptrunk_minimum_link
-
- ae_members = JuniperAeMembers # type: ignore[assignment]
+ ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
return ae_members # type: ignore[return-value]
@@ -92,7 +87,8 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv4_network)
iptrunk_ipv6_network: ipaddress.IPv6Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv6_network)
- @validator("tt_number", allow_reuse=True)
+ @field_validator("tt_number")
+ @classmethod
def validate_tt_number(cls, tt_number: str) -> str:
return validate_tt_number(tt_number)
@@ -100,8 +96,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
ae_members_side_a = initialize_ae_members(subscription, initial_user_input.dict(), 0)
class ModifyIptrunkSideAForm(FormPage):
- class Config:
- title = "Provide subscription details for side A of the trunk."
+ model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")
side_a_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn)
side_a_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface)
@@ -112,11 +107,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @validator("side_a_ae_members", allow_reuse=True)
+ @field_validator("side_a_ae_members")
+ @classmethod
def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
return validate_iptrunk_unique_interface(side_a_ae_members)
- @validator("side_a_ae_members", allow_reuse=True)
+ @field_validator("side_a_ae_members")
+ @classmethod
def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
vendor = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.vendor
return validate_interface_name_list(side_a_ae_members, vendor)
@@ -125,8 +122,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
ae_members_side_b = initialize_ae_members(subscription, initial_user_input.dict(), 1)
class ModifyIptrunkSideBForm(FormPage):
- class Config:
- title = "Provide subscription details for side B of the trunk."
+ model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")
side_b_node: str = ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn)
side_b_ae_iface: str = ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface)
@@ -137,11 +133,13 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
else []
)
- @validator("side_b_ae_members", allow_reuse=True)
+ @field_validator("side_b_ae_members")
+ @classmethod
def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
return validate_iptrunk_unique_interface(side_b_ae_members)
- @validator("side_b_ae_members", allow_reuse=True)
+ @field_validator("side_b_ae_members")
+ @classmethod
def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
vendor = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.vendor
return validate_interface_name_list(side_b_ae_members, vendor)
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index 7a2afe6c22f0e2a1f381c92734f0efd5a8dbbb25..bdd4e54183f411a2e87ebfa09632fa8d7bc14a43 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -16,7 +16,7 @@ from orchestrator.workflows.steps import (
unsync,
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import validator
+from pydantic import field_validator
from gso.products.product_blocks.iptrunk import IptrunkSideBlock
from gso.products.product_types.iptrunk import Iptrunk
@@ -41,7 +41,8 @@ def initial_input_form_generator() -> FormGenerator:
clean_up_ipam: bool = True
clean_up_netbox: bool = True
- @validator("tt_number", allow_reuse=True)
+ @field_validator("tt_number")
+ @classmethod
def validate_tt_number(cls, tt_number: str) -> str:
return validate_tt_number(tt_number)
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 289f201bd45b3064ad352c733a7c81aa1b7b7462..c10a9c77ee4129df3bfd366b6561afd9daa6274b 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -10,8 +10,8 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
from orchestrator.workflow import StepList, conditional, done, init, inputstep, 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_forms.core import ReadOnlyField
+from pydantic import ConfigDict, field_validator
+from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import RouterInactive, RouterProvisioning
@@ -39,8 +39,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Gather information about the new router from the operator."""
class CreateRouterForm(FormPage):
- class Config:
- title = product_name
+ model_config = ConfigDict(title=product_name)
tt_number: str
partner: str = ReadOnlyField("GEANT")
@@ -50,7 +49,8 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
ts_port: PortNumber
router_role: RouterRole
- @validator("hostname", allow_reuse=True)
+ @field_validator("hostname")
+ @classmethod
def hostname_must_be_available(cls, hostname: str, **kwargs: dict[str, Any]) -> str:
router_site = kwargs["values"].get("router_site")
if not router_site:
@@ -152,8 +152,7 @@ def prompt_reboot_router(subscription: RouterInactive) -> FormGenerator:
"""Wait for confirmation from an operator that the router has been rebooted."""
class RebootPrompt(FormPage):
- class Config:
- title = "Please reboot before continuing"
+ model_config = ConfigDict(title="Please reboot before continuing")
if subscription.router.router_site and subscription.router.router_site.site_ts_address:
info_label_1: Label = (
@@ -175,8 +174,7 @@ def prompt_console_login() -> FormGenerator:
"""Wait for confirmation from an operator that the router can be logged into."""
class ConsolePrompt(FormPage):
- class Config:
- title = "Verify local authentication"
+ model_config = ConfigDict(title="Verify local authentication")
info_label_1: Label = (
"Verify that you are able to log in to the router via the console using the admin account." # type: ignore[assignment]
@@ -193,8 +191,7 @@ def prompt_insert_in_ims() -> FormGenerator:
"""Wait for confirmation from an operator that the router has been inserted in IMS."""
class IMSPrompt(FormPage):
- class Config:
- title = "Update IMS mediation server"
+ model_config = ConfigDict(title="Update IMS mediation server")
info_label_1: Label = "Insert the router into IMS." # type: ignore[assignment]
info_label_2: Label = "Once this is done, press submit to continue the workflow." # type: ignore[assignment]
@@ -209,8 +206,7 @@ def prompt_insert_in_radius(subscription: RouterInactive) -> FormGenerator:
"""Wait for confirmation from an operator that the router has been inserted in RADIUS."""
class RadiusPrompt(FormPage):
- class Config:
- title = "Update RADIUS clients"
+ model_config = ConfigDict(title="Update RADIUS clients")
info_label_1: Label = (
f"Please go to https://kratos.geant.org/add_radius_client and add the {subscription.router.router_fqdn}" # type: ignore[assignment]
@@ -229,8 +225,7 @@ def prompt_start_new_checklist(subscription: RouterProvisioning) -> FormGenerato
oss_params = load_oss_params()
class SharepointPrompt(FormPage):
- class Config:
- title = "Start new checklist"
+ model_config = ConfigDict(title="Start new checklist")
info_label_1: Label = (
f"Visit {oss_params.SHAREPOINT.checklist_site_url} and start a new Sharepoint checklist for "
diff --git a/gso/workflows/router/modify_connection_strategy.py b/gso/workflows/router/modify_connection_strategy.py
index a3f5b5ae2f1f8cd0aa58d1d407d2daf28662c8a0..bc6be3ecc4b4a08605de17151ec7a14e740daf01 100644
--- a/gso/workflows/router/modify_connection_strategy.py
+++ b/gso/workflows/router/modify_connection_strategy.py
@@ -6,6 +6,7 @@ from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.workflow import StepList, done, init, 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
from gso.products.product_types.router import Router
from gso.utils.shared_enums import ConnectionStrategy
@@ -20,8 +21,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
class ModifyConnectionStrategyForm(FormPage):
- class Config:
- title = f"Modify the connection strategy of {subscription.router.router_fqdn}."
+ model_config = ConfigDict(title=f"Modify the connection strategy of {subscription.router.router_fqdn}.")
connection_strategy: ConnectionStrategy = current_connection_strategy
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index abef7da440b8024b578a7122be54f4440eb6764a..27085db5f361ddc4248ea2e1de507f30a3a6e03d 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -10,7 +10,7 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
from orchestrator.workflow import StepList, done, init, 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 root_validator
+from pydantic import ConfigDict, model_validator
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import Router
@@ -29,12 +29,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription = Router.from_subscription(subscription_id)
class AddBGPSessionForm(FormPage):
- class Config:
- title = f"Add {subscription.router.router_fqdn} to the iBGP mesh?"
+ model_config = ConfigDict(title=f"Add {subscription.router.router_fqdn} to the iBGP mesh?")
tt_number: str
- @root_validator(allow_reuse=True)
+ @model_validator()
+ @classmethod
def router_has_a_trunk(cls, values: dict[str, Any]) -> dict[str, Any]:
terminating_trunks = get_trunks_that_terminate_on_router(
subscription_id, SubscriptionLifecycle.PROVISIONING
@@ -198,8 +198,7 @@ def prompt_insert_in_radius() -> FormGenerator:
"""Wait for confirmation from an operator that the router has been inserted in RADIUS."""
class RADIUSPrompt(FormPage):
- class Config:
- title = "Please update RADIUS before continuing"
+ model_config = ConfigDict(title="Please update RADIUS before continuing")
info_label: Label = "Insert the router into RADIUS, and continue the workflow once this has been completed." # type: ignore[assignment]
@@ -213,8 +212,7 @@ def prompt_radius_login() -> FormGenerator:
"""Wait for confirmation from an operator that the router can be logged into using RADIUS."""
class RADIUSPrompt(FormPage):
- class Config:
- title = "Please check RADIUS before continuing"
+ model_config = ConfigDict(title="Please check RADIUS before continuing")
info_label: Label = "Log in to the router using RADIUS, and continue the workflow when this was successful." # type: ignore[assignment]
diff --git a/gso/workflows/site/create_site.py b/gso/workflows/site/create_site.py
index be9aab537c7fec01550b7f8009925b799a8c1fd9..6aefaba4b271315441641580621a22f5cc3b29e3 100644
--- a/gso/workflows/site/create_site.py
+++ b/gso/workflows/site/create_site.py
@@ -6,7 +6,8 @@ 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_forms.core import ReadOnlyField
+from pydantic import ConfigDict
+from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks import site as site_pb
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
@@ -19,8 +20,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Get input from the operator about the new site subscription."""
class CreateSiteForm(FormPage, BaseSiteValidatorModel):
- class Config:
- title = product_name
+ model_config = ConfigDict(title=product_name)
partner: str = ReadOnlyField("GEANT")
site_name: str
diff --git a/gso/workflows/site/modify_site.py b/gso/workflows/site/modify_site.py
index 15b549dbbcf7f357b5aebc28b885a998a18d9daa..5fe150169802ff810359eefbf44cbdefa74d7762 100644
--- a/gso/workflows/site/modify_site.py
+++ b/gso/workflows/site/modify_site.py
@@ -11,9 +11,8 @@ from orchestrator.workflows.steps import (
unsync,
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from pydantic import validator
-from pydantic.fields import ModelField
-from pydantic_forms.core import ReadOnlyField
+from pydantic import ConfigDict, field_validator
+from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks import site as site_pb
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
@@ -26,8 +25,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription = Site.from_subscription(subscription_id)
class ModifySiteForm(FormPage):
- class Config:
- title = "Modify Site"
+ model_config = ConfigDict(title="Modify Site")
site_name: str = ReadOnlyField(subscription.site.site_name)
site_city: str = subscription.site.site_city
@@ -40,18 +38,27 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
site_tier: site_pb.SiteTier = ReadOnlyField(subscription.site.site_tier)
site_ts_address: str | None = subscription.site.site_ts_address
- @validator("site_ts_address", allow_reuse=True)
+ @classmethod
+ @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
- @validator("site_internal_id", "site_bgp_community_id", allow_reuse=True)
- def validate_unique_fields(cls, value: str, field: ModelField) -> str | int:
- if value == getattr(subscription.site, field.name):
- return value
- return validate_site_fields_is_unique(field.name, value)
+ @classmethod
+ @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
+ return validate_site_fields_is_unique("site_internal_id", site_internal_id)
+
+ @classmethod
+ @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
+ return validate_site_fields_is_unique("site_bgp_community_id", site_bgp_community_id)
user_input = yield ModifySiteForm
diff --git a/gso/workflows/tasks/import_iptrunk.py b/gso/workflows/tasks/import_iptrunk.py
index 648d954f94ae57f9471826bba04402684508de81..330215bca0cd5664b0668584f42a0fa671f86a07 100644
--- a/gso/workflows/tasks/import_iptrunk.py
+++ b/gso/workflows/tasks/import_iptrunk.py
@@ -1,15 +1,18 @@
"""A creation workflow for adding an existing IP trunk to the service database."""
import ipaddress
+from typing import Annotated
from uuid import uuid4
from orchestrator import workflow
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, UniqueConstrainedList
+from orchestrator.forms.validators import Choice
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import AfterValidator, ConfigDict
+from pydantic_forms.validators import validate_unique_list
from gso.products import ProductName
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhysicalPortCapacity
@@ -29,14 +32,16 @@ def _generate_routers() -> dict[str, str]:
return routers
+LAGMemberList = Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
+
+
def initial_input_form_generator() -> FormGenerator:
"""Take all information passed to this workflow by the :term:`API` endpoint that was called."""
routers = _generate_routers()
router_enum = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
class CreateIptrunkForm(FormPage):
- class Config:
- title = "Import Iptrunk"
+ model_config = ConfigDict(title="Import Iptrunk")
partner: str
geant_s_sid: str
@@ -49,12 +54,12 @@ def initial_input_form_generator() -> FormGenerator:
side_a_node_id: router_enum # type: ignore[valid-type]
side_a_ae_iface: str
side_a_ae_geant_a_sid: str
- side_a_ae_members: UniqueConstrainedList[LAGMember]
+ side_a_ae_members: LAGMemberList
side_b_node_id: router_enum # type: ignore[valid-type]
side_b_ae_iface: str
side_b_ae_geant_a_sid: str
- side_b_ae_members: UniqueConstrainedList[LAGMember]
+ side_b_ae_members: LAGMemberList
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
diff --git a/gso/workflows/tasks/import_office_router.py b/gso/workflows/tasks/import_office_router.py
index 9168cdae0150a82a1893b6b3ceae450b5df542b5..99c2699c2a53c7221c6fca0669380a0f98fe18eb 100644
--- a/gso/workflows/tasks/import_office_router.py
+++ b/gso/workflows/tasks/import_office_router.py
@@ -8,6 +8,7 @@ from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import ConfigDict
from gso.products import ProductName
from gso.products.product_types import office_router
@@ -35,8 +36,7 @@ def initial_input_form_generator() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportOfficeRouter(FormPage):
- class Config:
- title = "Import an office router"
+ model_config = ConfigDict(title="Import an office router")
partner: str
office_router_site: str
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index c71ce26ee47a0e0929842d7261d7c0fd195d2e55..0e7a46139a76e52e3288d2cf1eb8c4cb7cd7dcce 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -8,6 +8,7 @@ from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import ConfigDict
from gso.products import ProductName
from gso.products.product_blocks import router as router_pb
@@ -38,8 +39,7 @@ def initial_input_form_generator() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportRouter(FormPage):
- class Config:
- title = "Import Router"
+ model_config = ConfigDict(title="Import Router")
partner: str
router_site: str
diff --git a/gso/workflows/tasks/import_site.py b/gso/workflows/tasks/import_site.py
index ff49808a5a86d1e73c6a741d74a75c4c7c233471..b2fa07a9a80efef5bdc7e8c2775cab4350760d26 100644
--- a/gso/workflows/tasks/import_site.py
+++ b/gso/workflows/tasks/import_site.py
@@ -7,6 +7,7 @@ from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import ConfigDict
from gso.products import ProductName
from gso.products.product_blocks.site import SiteTier
@@ -36,8 +37,7 @@ def generate_initial_input_form() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportSite(FormPage):
- class Config:
- title = "Import Site"
+ model_config = ConfigDict(title="Import Site")
site_name: str
site_city: str
diff --git a/gso/workflows/tasks/import_super_pop_switch.py b/gso/workflows/tasks/import_super_pop_switch.py
index 5f2796c2c2325ad439a0570db5154f57a0b435f1..dbc59b4c8cf8016076d184b537643cfe249f86c0 100644
--- a/gso/workflows/tasks/import_super_pop_switch.py
+++ b/gso/workflows/tasks/import_super_pop_switch.py
@@ -8,6 +8,7 @@ from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
from orchestrator.workflow import StepList, done, init, step
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import ConfigDict
from gso.products import ProductName
from gso.products.product_types import super_pop_switch
@@ -36,8 +37,7 @@ def initial_input_form_generator() -> FormGenerator:
"""Generate a form that is filled in using information passed through the :term:`API` endpoint."""
class ImportSuperPopSwitch(FormPage):
- class Config:
- title = "Import a Super PoP switch"
+ model_config = ConfigDict(title="Import a Super PoP switch")
partner: str
super_pop_switch_site: str
diff --git a/log.txt b/log.txt
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/requirements.txt b/requirements.txt
index 73bb46ff0e83e703d78bca7c58a728f95e7fa696..1083430c89b6b7da5d4e6ee2b485397a42349801 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-orchestrator-core==1.3.4
+orchestrator-core==2.1.2
requests==2.31.0
infoblox-client~=0.6.0
pycountry==22.3.5
diff --git a/test/auth/test_oidc_policy_helper.py b/test/auth/test_oidc_policy_helper.py
index 14af9f6b4ee55c5025aaef64414017f85a8f7513..f51b2dcfa8c0f1d0c715ff46cdef733c437edd5a 100644
--- a/test/auth/test_oidc_policy_helper.py
+++ b/test/auth/test_oidc_policy_helper.py
@@ -7,13 +7,7 @@ from httpx import AsyncClient, NetworkError, Response
from gso.auth.oidc_policy_helper import (
OIDCConfig,
- OIDCUser,
- OIDCUserModel,
- OPAResult,
- _evaluate_decision,
- _get_decision,
- _is_callback_step_endpoint,
- opa_decision,
+ OIDCUser, OIDCUserModel, OPAResult, opa_decision, _get_decision, _evaluate_decision, _is_callback_step_endpoint,
)
from gso.auth.settings import oauth2lib_settings