feat(platform): move platform from dcim to arch

backend/alembic/env.py

@@ -20,7 +20,14 @@
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-from src.db import Base  # noqa: E402
+from src.core.models.base import Base  # noqa: E402
+from src.features.admin.models import *
+from src.features.dcim.models import *
+from src.features.ipam.models import *
+from src.features.circuit.models import *
+from src.features.netconfig.models import *
+from src.features.arch.models import *
+from src.features.org.models import *
 
 target_metadata = Base.metadata
 

backend/pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
     "gunicorn>=21.2.0",
     "bcrypt>=4.0.1",
     "passlib>=1.7.4",
+    "cryptography>=42.0.8",
 ]
 readme = "README.md"
 requires-python = ">= 3.11"

backend/requirements-dev.lock

@@ -25,6 +25,8 @@ certifi==2023.11.17
     # via httpcore
     # via httpx
     # via sentry-sdk
+cffi==1.16.0
+    # via cryptography
 cfgv==3.4.0
     # via pre-commit
 click==8.1.7
@@ -33,6 +35,8 @@ click==8.1.7
     # via uvicorn
 coverage==7.4.0
     # via pytest-cov
+cryptography==42.0.8
+    # via netsight
 distlib==0.3.8
     # via virtualenv
 dnspython==2.5.0
@@ -108,6 +112,8 @@ platformdirs==4.1.0
 pluggy==1.3.0
     # via pytest
 pre-commit==3.6.0
+pycparser==2.22
+    # via cffi
 pydantic==2.5.3
     # via fastapi
     # via pydantic-extra-types

backend/requirements.lock

@@ -24,9 +24,13 @@ certifi==2023.11.17
     # via httpcore
     # via httpx
     # via sentry-sdk
+cffi==1.16.0
+    # via cryptography
 click==8.1.7
     # via typer
     # via uvicorn
+cryptography==42.0.8
+    # via netsight
 dnspython==2.5.0
     # via email-validator
 email-validator==2.1.0.post1
@@ -78,6 +82,8 @@ passlib==1.7.4
     # via netsight
 phonenumbers==8.13.27
     # via netsight
+pycparser==2.22
+    # via cffi
 pydantic==2.5.3
     # via fastapi
     # via pydantic-extra-types

backend/src/core/models/mixins/audit_log.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import TYPE_CHECKING
 
 from fastapi.encoders import jsonable_encoder
@@ -12,7 +13,6 @@
 from src.core.utils.context import orm_diff_ctx, request_id_ctx, user_ctx
 
 if TYPE_CHECKING:
-    from datetime import datetime
 
     from src.core.models.base import ModelT
     from src.features.admin.models import User
@@ -41,7 +41,7 @@ def get_object_change(obj: Mapper) -> dict:
 
 class AuditLog:
     id: Mapped[int_pk]
-    created_at: Mapped["datetime"] = mapped_column(DateTimeTZ, default=func.now())
+    created_at: Mapped[datetime] = mapped_column(DateTimeTZ, default=func.now())
     request_id: Mapped[str]
     action: Mapped[str] = mapped_column(String, nullable=False)
     diff: Mapped[dict | None] = mapped_column(JSON)

backend/src/features/arch/api.py

@@ -174,3 +174,44 @@ async def get_ip_role_auditlogs(self, id: int) -> ListT[AuditLog]:
         if not results:
             return ListT(count=0, results=None)
         return ListT(count=count, results=[AuditLog.model_validate(r) for r in results])
+
+
+@cbv(router)
+class PlatformAPI:
+    session: AsyncSession = Depends(get_session)
+    user: User = Depends(auth)
+    dto = BaseRepository(Platform)
+
+    @router.post("/platforms", operation_id="38e18494-e38c-4060-962f-64dfd37a61af")
+    async def create_platform(self, platform: schemas.PlatformCreate) -> IdResponse:
+        new_platform = await self.dto.create(self.session, platform)
+        return IdResponse(id=new_platform.id)
+
+    @router.put("/platforms/{id}", operation_id="a452765a-91b7-4b37-bca1-11864e1be028")
+    async def update_platform(self, id: int, platform: schemas.PlatformUpdate) -> IdResponse:
+        db_platform = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.update(self.session, db_platform, platform)
+        return IdResponse(id=id)
+
+    @router.get("/platforms/{id}", operation_id="9324bde0-600f-4470-98da-343fc498289c")
+    async def get_platform(self, id: int) -> schemas.Platform:
+        db_platform = await self.dto.get_one_or_404(self.session, id, undefer_load=True)
+        return schemas.Platform.model_validate(db_platform)
+
+    @router.get("/platforms", operation_id="d47d8d64-f8cc-4ddc-9db9-51d6a1f3b9e3")
+    async def get_platforms(self, q: schemas.PlatformQuery = Depends()) -> ListT[schemas.Platform]:
+        count, results = await self.dto.list_and_count(self.session, q)
+        return ListT(count=count, results=[schemas.Platform.model_validate(r) for r in results])
+
+    @router.delete("/platforms/{id}", operation_id="73a00be4-be83-4d24-a034-d36926bae8e1")
+    async def delete_platform(self, id: int) -> IdResponse:
+        db_platform = await self.dto.get_one_or_404(self.session, id)
+        await self.dto.delete(self.session, db_platform)
+        return IdResponse(id=id)
+
+    @router.get("/platforms/{id}/auditlogs", operation_id="62ca0f32-56b3-47bb-838f-20402c2bb1f4")
+    async def get_platform_audit_logs(self, id: int) -> ListT[AuditLog]:
+        count, results = await self.dto.get_audit_log(self.session, id)
+        if not results:
+            return ListT(count=0, results=None)
+        return ListT(count=count, results=[AuditLog.model_validate(r) for r in results])

backend/src/features/arch/models.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING
 
-from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.orm import Mapped, column_property, mapped_column, relationship
 
 from src.core.database.types import i18n_name, int_pk
 from src.core.models.base import Base
@@ -10,7 +10,7 @@
     from src.features.circuit.models import Circuit
     from src.features.ipam.models import VLAN, Prefix
 
-__all__ = ("RackRole", "DeviceRole", "IPRole", "CircuitType")
+__all__ = ("RackRole", "DeviceRole", "IPRole", "CircuitType", "Platform")
 
 
 class RackRole(Base, AuditLogMixin):
@@ -28,6 +28,8 @@ class DeviceRole(Base, AuditLogMixin):
     id: Mapped[int_pk]
     name: Mapped[i18n_name]
     slug: Mapped[str] = mapped_column(unique=True)
+    priority: Mapped[int]
+    abbreviation: Mapped[str | None]
     description: Mapped[str | None]
 
 
@@ -50,3 +52,34 @@ class CircuitType(Base, AuditLogMixin):
     slug: Mapped[str] = mapped_column(unique=True)
     description: Mapped[str | None]
     circuit: Mapped[list["Circuit"]] = relationship(back_populates="circuit_type")
+
+
+class Platform(Base, AuditLogMixin):
+    __tablename__ = "platform"
+    id: Mapped[int_pk]
+    name: Mapped[str] = mapped_column(unique=True)
+    slug: Mapped[str] = mapped_column(unique=True)
+    description: Mapped[str | None]
+    netmiko_driver: Mapped[str | None]
+
+
+    if TYPE_CHECKING:
+        device_count: Mapped[int]
+        device_type_count: Mapped[int]
+
+    @classmethod
+    def __declare_last__(cls) -> None:
+        from sqlalchemy import func, select
+
+        from src.features.dcim.models import Device, DeviceType
+
+        cls.device_count = column_property(
+            select(func.count(Device.id)).where(Device.platform_id == cls.id).scalar_subquery(),
+            deferred=True,
+        )
+        cls.device_type_count = column_property(
+            select(func.count(DeviceType.id))
+            .where(DeviceType.platform_id == cls.id)
+            .scalar_subquery(),
+            deferred=True,
+        )

backend/src/features/arch/schemas.py

@@ -1,7 +1,7 @@
 from fastapi import Query
 from pydantic import Field
 
-from src.features._types import AuditTime, BaseModel, I18nField, QueryParams
+from src.features._types import AuditTime, BaseModel, I18nField, NameStr, QueryParams
 
 
 class CircuitTypeCreate(BaseModel):
@@ -50,11 +50,13 @@ class DeviceRoleCreate(BaseModel):
     name: I18nField
     slug: str
     description: str | None = None
+    priority: int
 
 
 class DeviceRoleUpdate(BaseModel):
     name: I18nField | None = None
     slug: str | None = None
+    priority: int | None = None
 
 
 class DeviceRoleQuery(QueryParams):
@@ -87,3 +89,32 @@ class IPRole(IPRoleCreate, AuditTime):
     id: int
     prefix_count: int
     vlan_count: int
+
+
+class PlatformBase(BaseModel):
+    name: str
+    slug: str
+    description: str | None = None
+    netmiko_driver: str | None = None
+
+
+class PlatformCreate(PlatformBase):
+    name: NameStr
+    slug: NameStr
+
+
+class PlatformUpdate(PlatformCreate):
+    name: NameStr | None = None
+    slug: NameStr | None = None
+
+
+class PlatformQuery(QueryParams):
+    name: list[str] | None = Field(Query(default=[]))
+    slug: list[str] | None = Field(Query(default=[]))
+
+
+class Platform(PlatformBase, AuditTime):
+    id: int
+    device_type_count: int
+    device_count: int
+    textfsm_template_count: int

backend/src/features/circuit/models.py

@@ -5,21 +5,23 @@
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy_utils.types import ChoiceType
 
-from src.core.models.base import Base
-from src.core.models.mixins import AuditLogMixin
-from src.features._types import (
-    IPvAnyAddress,
-    IPvAnyInterface,
+from src.core.database.types import (
     PgIpAddress,
     PgIpInterface,
     date_optional,
     i18n_name,
     int_pk,
 )
+from src.core.models.base import Base
+from src.core.models.mixins import AuditLogMixin
+from src.features._types import IPvAnyAddress, IPvAnyInterface
 from src.features.consts import CircuitStatus
 
 if TYPE_CHECKING:
-    from src.db import ASN, CircuitContact, CircuitType, Device, Interface, Site
+    from src.features.arch.models import CircuitType
+    from src.features.dcim.models import Device, Interface
+    from src.features.ipam.models import ASN
+    from src.features.org.models import CircuitContact, Site
 
 __all__ = ("Circuit", "ISP", "ISPASN")
 

backend/src/features/consts.py

@@ -12,6 +12,7 @@ class ReservedRoleSlug(StrEnum):
 
 class SiteStatus(StrEnum):
     Planning = "Planning"
+    Deploying = "Deploying"
     Vlidating = "Validating"
     Active = "Active"
     Retired = "Retired"
@@ -35,12 +36,16 @@ class DeviceStatus(StrEnum):
     Offline = "Offline"
     Staged = "Staged"
     Inventory = "Inventory"
+    Scrapped = "Scrapped"
+
 
 
 class APStatus(StrEnum):
     Active = "Active"
     Offline = "Offline"
+    Staged = "Staged"
     Inventory = "Inventory"
+    Scrapped = "Scrapped"
 
 
 class APMode(StrEnum):

backend/src/features/dcim/api.py

@@ -171,46 +171,6 @@ async def get_device_type_audit_logs(self, id: int) -> ListT[AuditLog]:
         return ListT(count=count, results=[AuditLog.model_validate(r) for r in results])
 
 
-@cbv(router)
-class PlatformAPI:
-    session: AsyncSession = Depends(get_session)
-    user: User = Depends(auth)
-    dto = BaseRepository(Platform)
-
-    @router.post("/platforms", operation_id="38e18494-e38c-4060-962f-64dfd37a61af")
-    async def create_platform(self, platform: schemas.PlatformCreate) -> IdResponse:
-        new_platform = await self.dto.create(self.session, platform)
-        return IdResponse(id=new_platform.id)
-
-    @router.put("/platforms/{id}", operation_id="a452765a-91b7-4b37-bca1-11864e1be028")
-    async def update_platform(self, id: int, platform: schemas.PlatformUpdate) -> IdResponse:
-        db_platform = await self.dto.get_one_or_404(self.session, id)
-        await self.dto.update(self.session, db_platform, platform)
-        return IdResponse(id=id)
-
-    @router.get("/platforms/{id}", operation_id="9324bde0-600f-4470-98da-343fc498289c")
-    async def get_platform(self, id: int) -> schemas.Platform:
-        db_platform = await self.dto.get_one_or_404(self.session, id, undefer_load=True)
-        return schemas.Platform.model_validate(db_platform)
-
-    @router.get("/platforms", operation_id="d47d8d64-f8cc-4ddc-9db9-51d6a1f3b9e3")
-    async def get_platforms(self, q: schemas.PlatformQuery = Depends()) -> ListT[schemas.Platform]:
-        count, results = await self.dto.list_and_count(self.session, q)
-        return ListT(count=count, results=[schemas.Platform.model_validate(r) for r in results])
-
-    @router.delete("/platforms/{id}", operation_id="73a00be4-be83-4d24-a034-d36926bae8e1")
-    async def delete_platform(self, id: int) -> IdResponse:
-        db_platform = await self.dto.get_one_or_404(self.session, id)
-        await self.dto.delete(self.session, db_platform)
-        return IdResponse(id=id)
-
-    @router.get("/platforms/{id}/auditlogs", operation_id="62ca0f32-56b3-47bb-838f-20402c2bb1f4")
-    async def get_platform_audit_logs(self, id: int) -> ListT[AuditLog]:
-        count, results = await self.dto.get_audit_log(self.session, id)
-        if not results:
-            return ListT(count=0, results=None)
-        return ListT(count=count, results=[AuditLog.model_validate(r) for r in results])
-
 
 @cbv(router)
 class DeviceAPI: