107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import NetObjectModel, RelationModel
|
|
from app.schemas import (
|
|
EntityResponse, ListResponse, Relation,
|
|
RelationCreate, RelationUpdate,
|
|
)
|
|
from app.services.inventory import _to_relation
|
|
|
|
router = APIRouter(prefix='/relations', tags=['relations'])
|
|
|
|
|
|
@router.get('', response_model=ListResponse[Relation])
|
|
def list_relations(
|
|
source_id: str | None = None,
|
|
target_id: str | None = None,
|
|
layer: str | None = None,
|
|
type: str | None = Query(default=None),
|
|
min_confidence: float | None = None,
|
|
status: str | None = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
q = db.query(RelationModel)
|
|
if source_id:
|
|
q = q.filter(RelationModel.source_id == source_id)
|
|
if target_id:
|
|
q = q.filter(RelationModel.target_id == target_id)
|
|
if layer:
|
|
q = q.filter(RelationModel.layer == layer)
|
|
if type:
|
|
q = q.filter(RelationModel.type == type)
|
|
if min_confidence is not None:
|
|
q = q.filter(RelationModel.confidence >= min_confidence)
|
|
if status:
|
|
q = q.filter(RelationModel.status == status)
|
|
rows = q.all()
|
|
return ListResponse(count=len(rows), items=[_to_relation(r) for r in rows])
|
|
|
|
|
|
@router.get('/{relation_id}', response_model=EntityResponse[Relation])
|
|
def get_relation(relation_id: str, db: Session = Depends(get_db)):
|
|
row = db.query(RelationModel).filter(RelationModel.id == relation_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Relation not found')
|
|
return EntityResponse(item=_to_relation(row))
|
|
|
|
|
|
@router.post('', response_model=EntityResponse[Relation], status_code=201)
|
|
def create_relation(body: RelationCreate, db: Session = Depends(get_db)):
|
|
for fk, val in [('source_id', body.source_id), ('target_id', body.target_id)]:
|
|
if not db.query(NetObjectModel).filter(NetObjectModel.id == val).first():
|
|
raise HTTPException(status_code=422, detail=f'{fk} {val!r} not found')
|
|
now = datetime.now(timezone.utc)
|
|
row = RelationModel(
|
|
id=f'rel_{uuid4().hex[:8]}',
|
|
type=body.type,
|
|
layer=body.layer,
|
|
source_id=body.source_id,
|
|
target_id=body.target_id,
|
|
direction=body.direction,
|
|
source=body.source,
|
|
confidence=body.confidence,
|
|
status=body.status,
|
|
attributes_json=json.dumps(body.attributes),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return EntityResponse(item=_to_relation(row))
|
|
|
|
|
|
@router.patch('/{relation_id}', response_model=EntityResponse[Relation])
|
|
def update_relation(relation_id: str, body: RelationUpdate, db: Session = Depends(get_db)):
|
|
row = db.query(RelationModel).filter(RelationModel.id == relation_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Relation not found')
|
|
data = body.model_dump(exclude_unset=True)
|
|
for fk in ('source_id', 'target_id'):
|
|
if fk in data and data[fk]:
|
|
if not db.query(NetObjectModel).filter(NetObjectModel.id == data[fk]).first():
|
|
raise HTTPException(status_code=422, detail=f'{fk} {data[fk]!r} not found')
|
|
if 'attributes' in data:
|
|
data['attributes_json'] = json.dumps(data.pop('attributes'))
|
|
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_relation(row))
|
|
|
|
|
|
@router.delete('/{relation_id}', status_code=204)
|
|
def delete_relation(relation_id: str, db: Session = Depends(get_db)):
|
|
row = db.query(RelationModel).filter(RelationModel.id == relation_id).first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail='Relation not found')
|
|
db.delete(row)
|
|
db.commit()
|