olivier
28e7a3e3d2
update_vlan now checks for vlan_id conflicts (excluding the current record) before committing, matching the behaviour of create_vlan and preventing an unhandled IntegrityError 500. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
import ipaddress
|
|
import re
|
|
from typing import Optional, List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, field_validator
|
|
from sqlalchemy import nullsfirst
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
import models
|
|
|
|
router = APIRouter()
|
|
|
|
_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
|
|
|
|
|
|
class VlanCreate(BaseModel):
|
|
vlan_id: Optional[int] = None
|
|
name: str
|
|
cidr: Optional[str] = ""
|
|
color: str = "#4A90D9"
|
|
|
|
@field_validator("vlan_id")
|
|
@classmethod
|
|
def _vlan_id(cls, v: Optional[int]) -> Optional[int]:
|
|
if v is not None and not (1 <= v <= 4094):
|
|
raise ValueError("vlan_id must be between 1 and 4094")
|
|
return v
|
|
|
|
@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("cidr")
|
|
@classmethod
|
|
def _cidr(cls, v: Optional[str]) -> Optional[str]:
|
|
if v:
|
|
try:
|
|
ipaddress.ip_network(v, strict=False)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid CIDR notation: {v!r}")
|
|
return v
|
|
|
|
@field_validator("color")
|
|
@classmethod
|
|
def _color(cls, v: str) -> str:
|
|
if not _COLOR_RE.match(v):
|
|
raise ValueError("color must be a 6-digit hex color (e.g. #4A90D9)")
|
|
return v
|
|
|
|
|
|
class VlanOut(VlanCreate):
|
|
id: int
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
@router.get("/", response_model=List[VlanOut])
|
|
def list_vlans(db: Session = Depends(get_db)):
|
|
return db.query(models.Vlan).order_by(nullsfirst(models.Vlan.vlan_id)).all()
|
|
|
|
|
|
@router.post("/", response_model=VlanOut)
|
|
def create_vlan(vlan: VlanCreate, db: Session = Depends(get_db)):
|
|
if vlan.vlan_id is not None:
|
|
existing = db.query(models.Vlan).filter(models.Vlan.vlan_id == vlan.vlan_id).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail=f"VLAN {vlan.vlan_id} existe déjà")
|
|
db_vlan = models.Vlan(**vlan.model_dump())
|
|
db.add(db_vlan)
|
|
db.commit()
|
|
db.refresh(db_vlan)
|
|
return db_vlan
|
|
|
|
|
|
@router.put("/{vlan_pk}", response_model=VlanOut)
|
|
def update_vlan(vlan_pk: int, vlan: VlanCreate, db: Session = Depends(get_db)):
|
|
db_vlan = db.query(models.Vlan).filter(models.Vlan.id == vlan_pk).first()
|
|
if not db_vlan:
|
|
raise HTTPException(status_code=404, detail="VLAN introuvable")
|
|
if vlan.vlan_id is not None:
|
|
conflict = (
|
|
db.query(models.Vlan)
|
|
.filter(models.Vlan.vlan_id == vlan.vlan_id, models.Vlan.id != vlan_pk)
|
|
.first()
|
|
)
|
|
if conflict:
|
|
raise HTTPException(status_code=400, detail=f"VLAN {vlan.vlan_id} existe déjà")
|
|
for k, v in vlan.model_dump().items():
|
|
setattr(db_vlan, k, v)
|
|
db.commit()
|
|
db.refresh(db_vlan)
|
|
return db_vlan
|
|
|
|
|
|
@router.delete("/{vlan_pk}")
|
|
def delete_vlan(vlan_pk: int, db: Session = Depends(get_db)):
|
|
db_vlan = db.query(models.Vlan).filter(models.Vlan.id == vlan_pk).first()
|
|
if not db_vlan:
|
|
raise HTTPException(status_code=404, detail="VLAN introuvable")
|
|
|
|
# Collect devices with an interface in this VLAN
|
|
ifaces = (
|
|
db.query(models.DeviceInterface)
|
|
.filter(models.DeviceInterface.vlan_id == vlan_pk)
|
|
.all()
|
|
)
|
|
device_ids = {i.device_id for i in ifaces}
|
|
|
|
for device_id in device_ids:
|
|
device = db.query(models.Device).filter(models.Device.id == device_id).first()
|
|
if device and not device.is_gateway and not device.is_livebox:
|
|
db.delete(device) # cascade deletes all its interfaces
|
|
|
|
# Gateway/livebox interfaces in this VLAN will be SET NULL by SQLAlchemy
|
|
db.delete(db_vlan)
|
|
db.commit()
|
|
return {"ok": True}
|