stupid-simple-network-inventory/backend/routers/devices.py

import ipaddress
import re
from typing import Optional, List
from urllib.parse import urlparse

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, field_validator
from sqlalchemy.orm import Session

from database import get_db
import models

router = APIRouter()

_VALID_TYPES = {
    "server", "switch", "router", "nas", "gateway", "livebox", "access_point",
    "camera", "temperature", "sensor", "hub", "smart_plug", "alarm", "light",
    "doorbell", "desktop", "laptop", "smart_tv", "printer", "smartphone", "other",
}
_VALID_VIRT_TYPES = {None, "baremetal", "lxc", "qemu"}


class InterfaceCreate(BaseModel):
    name: str = "eth0"
    ip_address: Optional[str] = ""
    vlan_id: Optional[int] = None
    is_upstream: bool = False

    @field_validator("name")
    @classmethod
    def _name(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("Interface name cannot be empty")
        if len(v) > 50:
            raise ValueError("Interface name too long (max 50 characters)")
        return v

    @field_validator("ip_address")
    @classmethod
    def _ip(cls, v: Optional[str]) -> Optional[str]:
        if v:
            try:
                ipaddress.ip_address(v)
            except ValueError:
                raise ValueError(f"Invalid IP address: {v!r}")
        return v


class InterfaceOut(InterfaceCreate):
    id: int
    device_id: int

    model_config = {"from_attributes": True}


class DeviceCreate(BaseModel):
    name: str
    type: str = "other"
    description: str = ""
    is_gateway: bool = False
    is_livebox: bool = False
    virt_type: Optional[str] = None
    url: Optional[str] = None
    interfaces: List[InterfaceCreate] = []

    @field_validator("name")
    @classmethod
    def _name(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("name cannot be empty")
        if len(v) > 100:
            raise ValueError("name too long (max 100 characters)")
        return v

    @field_validator("description")
    @classmethod
    def _description(cls, v: str) -> str:
        if len(v) > 500:
            raise ValueError("description too long (max 500 characters)")
        return v

    @field_validator("type")
    @classmethod
    def _type(cls, v: str) -> str:
        if v not in _VALID_TYPES:
            raise ValueError(f"Invalid type: {v!r}. Must be one of: {sorted(_VALID_TYPES)}")
        return v

    @field_validator("virt_type")
    @classmethod
    def _virt_type(cls, v: Optional[str]) -> Optional[str]:
        if v not in _VALID_VIRT_TYPES:
            raise ValueError(f"Invalid virt_type: {v!r}. Must be one of: baremetal, lxc, qemu")
        return v

    @field_validator("url")
    @classmethod
    def _url(cls, v: Optional[str]) -> Optional[str]:
        if v:
            parsed = urlparse(v)
            if parsed.scheme not in ("http", "https") or not parsed.netloc:
                raise ValueError("url must be a valid http or https URL")
        return v


class DeviceOut(BaseModel):
    id: int
    name: str
    type: str
    description: str
    is_gateway: bool
    is_livebox: bool
    virt_type: Optional[str] = None
    url: Optional[str] = None
    interfaces: List[InterfaceOut] = []

    model_config = {"from_attributes": True}


@router.get("/", response_model=List[DeviceOut])
def list_devices(db: Session = Depends(get_db)):
    return db.query(models.Device).order_by(models.Device.name).all()


@router.post("/", response_model=DeviceOut)
def create_device(device: DeviceCreate, db: Session = Depends(get_db)):
    db_device = models.Device(
        name=device.name,
        type=device.type,
        description=device.description,
        is_gateway=device.is_gateway,
        is_livebox=device.is_livebox,
        virt_type=device.virt_type,
        url=device.url,
    )
    db.add(db_device)
    db.flush()
    for iface in device.interfaces:
        db.add(models.DeviceInterface(device_id=db_device.id, **iface.model_dump()))
    db.commit()
    db.refresh(db_device)
    return db_device


@router.put("/{device_id}", response_model=DeviceOut)
def update_device(device_id: int, device: DeviceCreate, db: Session = Depends(get_db)):
    db_device = db.query(models.Device).filter(models.Device.id == device_id).first()
    if not db_device:
        raise HTTPException(status_code=404, detail="Équipement introuvable")
    db_device.name = device.name
    db_device.type = device.type
    db_device.description = device.description
    db_device.is_gateway = device.is_gateway
    db_device.is_livebox = device.is_livebox
    db_device.virt_type = device.virt_type
    db_device.url = device.url
    db.query(models.DeviceInterface).filter(
        models.DeviceInterface.device_id == device_id
    ).delete()
    for iface in device.interfaces:
        db.add(models.DeviceInterface(device_id=device_id, **iface.model_dump()))
    db.commit()
    db.refresh(db_device)
    return db_device


@router.delete("/{device_id}")
def delete_device(device_id: int, db: Session = Depends(get_db)):
    db_device = db.query(models.Device).filter(models.Device.id == device_id).first()
    if not db_device:
        raise HTTPException(status_code=404, detail="Équipement introuvable")
    db.delete(db_device)
    db.commit()
    return {"ok": True}