121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
import re
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import NetObjectModel
|
|
from app.schemas import EntityResponse, ListResponse, NetObject, NetObjectCreate, NetObjectUpdate
|
|
from app.services.inventory import _to_net_object
|
|
|
|
router = APIRouter(prefix='/objects', tags=['objects'])
|
|
|
|
|
|
def _slugify(name: str) -> str:
|
|
slug = name.lower().strip()
|
|
slug = re.sub(r'[^\w\s-]', '', slug)
|
|
slug = re.sub(r'[\s_-]+', '-', slug)
|
|
return slug.strip('-')
|
|
|
|
|
|
@router.get('', response_model=ListResponse[NetObject])
|
|
def list_objects(db: Session = Depends(get_db)):
|
|
rows = db.query(NetObjectModel).all()
|
|
return ListResponse(count=len(rows), items=[_to_net_object(r) for r in rows])
|
|
|
|
|
|
@router.get('/{object_id}', response_model=EntityResponse[NetObject])
|
|
def get_object(object_id: str, db: Session = Depends(get_db)):
|
|
row = db.query(NetObjectModel).filter(NetObjectModel.id == object_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Object not found')
|
|
return EntityResponse(item=_to_net_object(row))
|
|
|
|
|
|
@router.post('', response_model=EntityResponse[NetObject], status_code=201)
|
|
def create_object(body: NetObjectCreate, db: Session = Depends(get_db)):
|
|
if body.parent_id:
|
|
parent = db.query(NetObjectModel).filter(NetObjectModel.id == body.parent_id).first()
|
|
if not parent:
|
|
raise HTTPException(status_code=422, detail=f'parent_id {body.parent_id!r} not found')
|
|
now = datetime.now(timezone.utc)
|
|
obj = NetObjectModel(
|
|
id=f'obj_{uuid4().hex[:8]}',
|
|
name=body.name,
|
|
slug=_slugify(body.name),
|
|
type=body.type,
|
|
parent_id=body.parent_id,
|
|
status=body.status,
|
|
vendor=body.vendor,
|
|
model=body.model,
|
|
serial=body.serial,
|
|
notes=body.notes,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return EntityResponse(item=_to_net_object(obj))
|
|
|
|
|
|
@router.put('/{object_id}', response_model=EntityResponse[NetObject])
|
|
def replace_object(object_id: str, body: NetObjectCreate, db: Session = Depends(get_db)):
|
|
row = db.query(NetObjectModel).filter(NetObjectModel.id == object_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Object not found')
|
|
if body.parent_id and body.parent_id != row.parent_id:
|
|
parent = db.query(NetObjectModel).filter(NetObjectModel.id == body.parent_id).first()
|
|
if not parent:
|
|
raise HTTPException(status_code=422, detail=f'parent_id {body.parent_id!r} not found')
|
|
row.name = body.name
|
|
row.slug = _slugify(body.name)
|
|
row.type = body.type
|
|
row.parent_id = body.parent_id
|
|
row.status = body.status
|
|
row.vendor = body.vendor
|
|
row.model = body.model
|
|
row.serial = body.serial
|
|
row.notes = body.notes
|
|
row.updated_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return EntityResponse(item=_to_net_object(row))
|
|
|
|
|
|
@router.patch('/{object_id}', response_model=EntityResponse[NetObject])
|
|
def update_object(object_id: str, body: NetObjectUpdate, db: Session = Depends(get_db)):
|
|
row = db.query(NetObjectModel).filter(NetObjectModel.id == object_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Object not found')
|
|
data = body.model_dump(exclude_unset=True)
|
|
if 'parent_id' in data and data['parent_id']:
|
|
parent = db.query(NetObjectModel).filter(NetObjectModel.id == data['parent_id']).first()
|
|
if not parent:
|
|
raise HTTPException(status_code=422, detail=f'parent_id {data["parent_id"]!r} not found')
|
|
if 'name' in data:
|
|
row.slug = _slugify(data['name'])
|
|
for field, value in data.items():
|
|
setattr(row, field, value)
|
|
row.updated_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return EntityResponse(item=_to_net_object(row))
|
|
|
|
|
|
@router.delete('/{object_id}', status_code=204)
|
|
def delete_object(object_id: str, db: Session = Depends(get_db)):
|
|
row = db.query(NetObjectModel).filter(NetObjectModel.id == object_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Object not found')
|
|
children = db.query(NetObjectModel).filter(NetObjectModel.parent_id == object_id).count()
|
|
if children:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f'Cannot delete: object has {children} child object(s). Remove or re-parent them first.',
|
|
)
|
|
db.delete(row)
|
|
db.commit()
|