upgrade orchestrator-core to v2

.gitlab-ci.yml

@@ -12,7 +12,7 @@ run-tox-pipeline:
   stage: tox
   tags:
     - docker-executor
-  image: python:3.11
+  image: python:3.12
 
   services:
     - postgres:15.4

Dockerfile

-FROM python:3.11-alpine
+FROM python:3.12-alpine
 WORKDIR /app
 
 ARG ARTIFACT_VERSION

docs/source/quickstart.rst

@@ -4,15 +4,15 @@ Quickstart
 Development environment and dependencies
 ----------------------------------------
 
-- Install python 3.11 if you do not have it already:
+- Install python 3.12 if you do not have it already:
     - ``add-apt-repository ppa:deadsnakes/ppa``
-    - ``apt install python3.11 python3.11-distutils``
+    - ``apt install python3.12 python3.12-distutils``
 - Follow Steps 1 and 2 from here to install dependencies and setup DB:
   `<https://workfloworchestrator.org/orchestrator-core/workshops/beginner/debian/>`_
 - To install the orchestrator GUI, you can follow the steps 5 and 6 from the previous link.
 - Create a virtual environment:
     - ``source /usr/share/virtualenvwrapper/virtualenvwrapper.sh``
-    - ``mkvirtualenv --python python3.11 gso``
+    - ``mkvirtualenv --python python3.12 gso``
 - To use the virtual environment:
     - ``source /usr/share/virtualenvwrapper/virtualenvwrapper.sh``
     - ``workon gso``
@@ -25,7 +25,7 @@ Do all this inside the virtual environment.
 - Clone this repository
 - ``pip install -r requirements.txt``
     - If you get an error because you pip version is too old, run this:
-      ``curl -sS https://bootstrap.pypa.io/get-pip.py | python3.11``
+      ``curl -sS https://bootstrap.pypa.io/get-pip.py | python3.12``
 - ``pip install -e .``
 - Create an ``oss-params.json`` based on the ``oss-params-example.json`` file inside ``/gso``.
 - Export the oss-params file: ``export OSS_PARAMS_FILENAME="/path/to/oss-params.json"``

gso/__init__.py

@@ -16,6 +16,7 @@ from gso.middlewares import ModifyProcessEndpointResponse
 def init_gso_app() -> OrchestratorCore:
     """Initialise the :term:`GSO` app."""
     app = OrchestratorCore(base_settings=app_settings)
+    # app.register_graphql() # TODO: uncomment this line when the GUI V2 is ready
     app.include_router(api_router, prefix="/api")
     app.add_middleware(ModifyProcessEndpointResponse)
     return app
@@ -28,7 +29,7 @@ def init_worker_app() -> OrchestratorCore:
 
 def init_cli_app() -> typer.Typer:
     """Initialise :term:`GSO` as a CLI application."""
-    from gso.cli import imports, netbox  # noqa: PLC0415
+    from gso.cli import imports, netbox
 
     cli_app.add_typer(imports.app, name="import-cli")
     cli_app.add_typer(netbox.app, name="netbox-cli")

gso/auth/oidc_policy_helper.py

@@ -252,7 +252,7 @@ class OIDCUser(HTTPBearer):
             return
 
         response = await async_request.get(self.openid_url + "/.well-known/openid-configuration")
-        self.openid_config = OIDCConfig.parse_obj(response.json())
+        self.openid_config = OIDCConfig.model_validate(response.json())
 
     async def userinfo(self, async_request: AsyncClient, token: str) -> OIDCUserModel:
         """Get the userinfo from the openid server.

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):

gso/middlewares.py

@@ -93,9 +93,9 @@ class ModifyProcessEndpointResponse(BaseHTTPMiddleware):
                     if callback_result and isinstance(callback_result, str):
                         callback_result = json.loads(callback_result)
                     if callback_result.get("output") and len(callback_result["output"]) > max_output_length:
-                        callback_result[
-                            "output"
-                        ] = f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+                        callback_result["output"] = (
+                            f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+                        )
                     step["state"]["callback_result"] = callback_result
         except (AttributeError, KeyError, TypeError):
             pass

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
 

gso/migrations/versions/2024-04-20_1ec810b289c0_add_orchestrator_2_1_2_migrations.py

+"""remove subscription cancellation workflow.
+
+Revision ID: 1ec810b289c0
+Revises:
+Create Date: 2024-04-02 10:21:08.539591
+
+"""
+from alembic import op
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+# revision identifiers, used by Alembic.
+revision = '1ec810b289c0'
+down_revision = '393acfa175c0'
+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.
+
+new_workflows = [
+    {
+        "name": "import_site",
+        "target": "SYSTEM",
+        "description": "Import a site without provisioning it.",
+        "product_type": "Site"
+    },
+    {
+        "name": "import_router",
+        "target": "SYSTEM",
+        "description": "Import a router without provisioning it.",
+        "product_type": "Router"
+    },
+    {
+        "name": "import_iptrunk",
+        "target": "SYSTEM",
+        "description": "Import an IP trunk without provisioning it.",
+        "product_type": "Iptrunk"
+    },
+    {
+        "name": "import_super_pop_switch",
+        "target": "SYSTEM",
+        "description": "Import a Super PoP switch without provisioning it.",
+        "product_type": "SuperPopSwitch"
+    },
+    {
+        "name": "import_office_router",
+        "target": "SYSTEM",
+        "description": "Import an office router without provisioning it.",
+        "product_type": "OfficeRouter"
+    },
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/oss-params-example.json

@@ -45,7 +45,7 @@
     },
     "LT_IAS": {
       "V4": {"containers": ["10.255.255.0/24"], "networks": [], "mask": 31},
-      "V6": {"containers": ["dead:beef:cc::/48"], "networks": [], "mask": 126},
+      "V6": {"containers": [ "2001:798:1::/48"], "networks": [], "mask": 126},
       "domain_name": ".geantip",
       "dns_view": "default",
       "network_view": "default"
@@ -73,6 +73,7 @@
   "PROVISIONING_PROXY": {
     "scheme": "https",
     "api_base": "localhost:44444",
+    "auth": "Bearer <token>",
     "api_version": 1123
   },
   "CELERY": {

gso/products/product_blocks/iptrunk.py

 """IP trunk product block that has all parameters of a subscription throughout its lifecycle."""
 
 import ipaddress
-from typing import TypeVar
+from typing import Annotated
 
-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."""
+LAGMemberList = Annotated[list[T], AfterValidator(validate_unique_list), Len(min_length=0)]
+IptrunkSides = Annotated[list[T], AfterValidator(validate_unique_list), Len(min_length=2, max_length=2)]
 
 
 class IptrunkInterfaceBlockInactive(
@@ -65,13 +64,6 @@ class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[Subscr
     interface_description: str | None = None
 
 
-class IptrunkSides(UniqueConstrainedList[T_co]):  # type: ignore[type-var]
-    """A list of IP trunk interfaces that make up one side of a link."""
-
-    min_items = 2
-    max_items = 2
-
-
 class IptrunkSideBlockInactive(
     ProductBlockModel,
     lifecycle=[SubscriptionLifecycle.INITIAL],
@@ -91,7 +83,7 @@ class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[Subscrip
     iptrunk_side_node: RouterBlockProvisioning
     iptrunk_side_ae_iface: str | None = None
     iptrunk_side_ae_geant_a_sid: str | None = None
-    iptrunk_side_ae_members: LAGMemberList[IptrunkInterfaceBlockProvisioning]
+    iptrunk_side_ae_members: LAGMemberList[IptrunkInterfaceBlockProvisioning]  # type: ignore[assignment]
 
 
 class IptrunkSideBlock(IptrunkSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -100,7 +92,7 @@ class IptrunkSideBlock(IptrunkSideBlockProvisioning, lifecycle=[SubscriptionLife
     iptrunk_side_node: RouterBlock
     iptrunk_side_ae_iface: str | None = None
     iptrunk_side_ae_geant_a_sid: str | None = None
-    iptrunk_side_ae_members: LAGMemberList[IptrunkInterfaceBlock]
+    iptrunk_side_ae_members: LAGMemberList[IptrunkInterfaceBlock]  # type: ignore[assignment]
 
 
 class IptrunkBlockInactive(
@@ -132,7 +124,7 @@ class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLife
     iptrunk_isis_metric: int | None = None
     iptrunk_ipv4_network: ipaddress.IPv4Network | None = None
     iptrunk_ipv6_network: ipaddress.IPv6Network | None = None
-    iptrunk_sides: IptrunkSides[IptrunkSideBlockProvisioning]
+    iptrunk_sides: IptrunkSides[IptrunkSideBlockProvisioning]  # type: ignore[assignment]
 
 
 class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -155,4 +147,4 @@ class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
     #:  The IPv6 network used for this trunk.
     iptrunk_ipv6_network: ipaddress.IPv6Network
     #:  The two sides that the trunk is connected to.
-    iptrunk_sides: IptrunkSides[IptrunkSideBlock]
+    iptrunk_sides: IptrunkSides[IptrunkSideBlock]  # type: ignore[assignment]

gso/products/product_blocks/lan_switch_interconnect.py

@@ -63,7 +63,7 @@ class LanSwitchInterconnectRouterSideBlockProvisioning(
 
     node: RouterBlockProvisioning
     ae_iface: str | None = None
-    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]  # type: ignore[assignment]
 
 
 class LanSwitchInterconnectRouterSideBlock(
@@ -73,7 +73,7 @@ class LanSwitchInterconnectRouterSideBlock(
 
     node: RouterBlock
     ae_iface: str
-    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]  # type: ignore[assignment]
 
 
 class LanSwitchInterconnectSwitchSideBlockInactive(
@@ -95,7 +95,7 @@ class LanSwitchInterconnectSwitchSideBlockProvisioning(
 
     node: SwitchBlockProvisioning
     ae_iface: str | None = None
-    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]  # type: ignore[assignment]
 
 
 class LanSwitchInterconnectSwitchSideBlock(
@@ -105,7 +105,7 @@ class LanSwitchInterconnectSwitchSideBlock(
 
     node: SwitchBlock
     ae_iface: str
-    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]
+    ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]  # type: ignore[assignment]
 
 
 class LanSwitchInterconnectBlockInactive(

gso/products/product_blocks/office_router.py

 """Product block for :class:`office router` products."""
 
-import ipaddress
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 
@@ -10,7 +8,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
 
 
 class OfficeRouterBlockInactive(
@@ -22,8 +20,8 @@ class OfficeRouterBlockInactive(
 
     office_router_fqdn: str | None = None
     office_router_ts_port: PortNumber | None = None
-    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_lo_ipv4_address: IPv4AddressType | None = None
+    office_router_lo_ipv6_address: IPv6AddressType | None = None
     office_router_site: SiteBlockInactive | None
     vendor: Vendor | None = None
 
@@ -33,8 +31,8 @@ class OfficeRouterBlockProvisioning(OfficeRouterBlockInactive, lifecycle=[Subscr
 
     office_router_fqdn: str | None = None
     office_router_ts_port: PortNumber | None = None
-    office_router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    office_router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    office_router_lo_ipv4_address: IPv4AddressType | None = None
+    office_router_lo_ipv6_address: IPv6AddressType | None = None
     office_router_site: SiteBlockProvisioning | None
     vendor: Vendor | None = None
 
@@ -47,9 +45,9 @@ class OfficeRouterBlock(OfficeRouterBlockProvisioning, lifecycle=[SubscriptionLi
     #:  The port of the terminal server that this office router is connected to. Used to offer out of band access.
     office_router_ts_port: PortNumber
     #:  The IPv4 loopback address of the office router.
-    office_router_lo_ipv4_address: ipaddress.IPv4Address
+    office_router_lo_ipv4_address: IPv4AddressType
     #:  The IPv6 loopback address of the office router.
-    office_router_lo_ipv6_address: ipaddress.IPv6Address
+    office_router_lo_ipv6_address: IPv6AddressType
     #:  The :class:`Site` that this office router resides in. Both physically and computationally.
     office_router_site: SiteBlock
     #:  The vendor of an office router. Defaults to Juniper.

gso/products/product_blocks/pop_vlan.py

 """Pop VLAN product block that has all parameters of a subscription throughout its lifecycle."""
 
 from ipaddress import IPv4Network, IPv6Network
-from typing import TypeVar
+from typing import Annotated, TypeVar
 
 from orchestrator.domain.base import ProductBlockModel
-from orchestrator.forms.validators import UniqueConstrainedList
 from orchestrator.types import SubscriptionLifecycle
+from pydantic import AfterValidator
 from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
 
 from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectBlock,
@@ -14,7 +15,7 @@ from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectBlockProvisioning,
 )
 
-T_co = TypeVar("T_co", covariant=True)
+T = TypeVar("T")
 
 
 class LayerPreference(strEnum):
@@ -24,8 +25,7 @@ class LayerPreference(strEnum):
     L3 = "L3"
 
 
-class PortList(UniqueConstrainedList[T_co]):  # type: ignore[type-var]
-    """A list of ports."""
+PortList = Annotated[list[T], AfterValidator(validate_unique_list)]
 
 
 class PopVlanPortBlockInactive(
@@ -92,7 +92,7 @@ class PopVlanBlock(PopVlanBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
     #: The LAN Switch Interconnect that this Pop VLAN is connected to.
     lan_switch_interconnect: LanSwitchInterconnectBlock
     #: The ports of the Pop VLAN.
-    ports: PortList[PopVlanPortBlock]
+    ports: PortList[PopVlanPortBlock]  # type: ignore[assignment]
     #: The level of the layer preference for the Pop VLAN (L2 or L3).
     layer_preference: LayerPreference
     #: IPv4 network for the Pop VLAN if layer preference is L3.

gso/products/product_blocks/router.py

 """Product block for :class:`Router` products."""
 
-import ipaddress
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
 
@@ -10,7 +8,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
 
 
 class RouterRole(strEnum):
@@ -31,8 +29,8 @@ class RouterBlockInactive(
     router_fqdn: str | None = None
     router_ts_port: PortNumber | None = None
     router_access_via_ts: bool | None = None
-    router_lo_ipv4_address: ipaddress.IPv4Address | None = None
-    router_lo_ipv6_address: ipaddress.IPv6Address | None = None
+    router_lo_ipv4_address: IPv4AddressType | None = None
+    router_lo_ipv6_address: IPv6AddressType | None = None
     router_lo_iso_address: str | None = None
     router_role: RouterRole | None = None
     router_site: SiteBlockInactive | None
@@ -45,8 +43,8 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
     router_fqdn: str
     router_ts_port: PortNumber
     router_access_via_ts: bool
-    router_lo_ipv4_address: ipaddress.IPv4Address
-    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_ipv4_address: IPv4AddressType
+    router_lo_ipv6_address: IPv6AddressType
     router_lo_iso_address: str
     router_role: RouterRole
     router_site: SiteBlockProvisioning
@@ -63,9 +61,9 @@ class RouterBlock(RouterBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTI
     #:  Whether this router should be accessed through the terminal server, or through its loopback address.
     router_access_via_ts: bool
     #:  The IPv4 loopback address of the router.
-    router_lo_ipv4_address: ipaddress.IPv4Address
+    router_lo_ipv4_address: IPv4AddressType
     #:  The IPv6 loopback address of the router.
-    router_lo_ipv6_address: ipaddress.IPv6Address
+    router_lo_ipv6_address: IPv6AddressType
     #:  The :term:`ISO` :term:`NET` of the router, used for :term:`ISIS` support.
     router_lo_iso_address: str
     #:  The role of the router, which can be any of the values defined in :class:`RouterRole`.

gso/products/product_blocks/site.py

 """The product block that describes a site subscription."""
 
 import re
+from typing import Annotated
 
+from annotated_types import doc
 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 +22,53 @@ 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:
+    """Validate a latitude coordinate."""
     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)):
+        msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
+        raise ValueError(msg)
+    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:
+    """Validate a longitude coordinate."""
     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)):
+        msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
+        raise ValueError(msg)
+
+    return v
+
+
+LatitudeCoordinate = Annotated[
+    float,
+    Field(
+        ge=-90,
+        le=90,
+    ),
+    AfterValidator(validate_latitude),
+    doc(
+        "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."
+    ),
+]
+
+LongitudeCoordinate = Annotated[
+    float,
+    Field(
+        ge=-180,
+        le=180,
+    ),
+    AfterValidator(validate_longitude),
+    doc(
+            "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."
+    ),
+]
 
 
 class SiteBlockInactive(

gso/products/product_blocks/super_pop_switch.py

 """Product block for :class:`Super PoP Switch` products."""
 
-import ipaddress
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 
@@ -10,7 +8,7 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
 
 
 class SuperPopSwitchBlockInactive(
@@ -22,7 +20,7 @@ class SuperPopSwitchBlockInactive(
 
     super_pop_switch_fqdn: str | None = None
     super_pop_switch_ts_port: PortNumber | None = None
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_mgmt_ipv4_address: IPv4AddressType | None = None
     super_pop_switch_site: SiteBlockInactive | None
     vendor: Vendor | None = None
 
@@ -32,7 +30,7 @@ class SuperPopSwitchBlockProvisioning(SuperPopSwitchBlockInactive, lifecycle=[Su
 
     super_pop_switch_fqdn: str | None = None
     super_pop_switch_ts_port: PortNumber | None = None
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address | None = None
+    super_pop_switch_mgmt_ipv4_address: IPv4AddressType | None = None
     super_pop_switch_site: SiteBlockProvisioning | None
     vendor: Vendor | None = None
 
@@ -45,7 +43,7 @@ class SuperPopSwitchBlock(SuperPopSwitchBlockProvisioning, lifecycle=[Subscripti
     #:  The port of the terminal server that this Super PoP switch is connected to. Used to offer out of band access.
     super_pop_switch_ts_port: PortNumber
     #:  The IPv4 management address of the Super PoP switch.
-    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address
+    super_pop_switch_mgmt_ipv4_address: IPv4AddressType
     #:  The :class:`Site` that this Super PoP switch resides in. Both physically and computationally.
     super_pop_switch_site: SiteBlock
     #:  The vendor of a Super PoP switch. Defaults to Juniper.

gso/schedules/scheduling.py

@@ -29,6 +29,7 @@ def scheduler(
 
     All time units can be specified with lists of numbers or crontab pattern strings for advanced scheduling.
     All specified time parts (minute, hour, day, etc.) must align for a task to run.
+
     """
 
     def decorator(task_func: Callable) -> Callable:

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
 
@@ -14,7 +14,7 @@ class PartnerCreate(BaseModel):
     partner_id: str = Field(default_factory=lambda: str(uuid4()))
     name: str
     email: EmailStr | None = None
-    as_number: str | None = Field(None, unique=True)
+    as_number: str | None = None
     as_set: str | None = None
     route_set: str | None = None
     black_listed_as_sets: list[str] | None = None
@@ -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)