172 lines
4.5 KiB
Python
172 lines
4.5 KiB
Python
from typing import Annotated, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
|
|
from app.core.database import get_db
|
|
from app.models import Allarme, Utente, Sito
|
|
from app.schemas.allarme import AllarmeResponse, AllarmeList, AllarmeUpdate
|
|
from app.api.auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/allarmi", tags=["Allarmi"])
|
|
|
|
|
|
@router.get("", response_model=AllarmeList)
|
|
async def get_allarmi(
|
|
current_user: Annotated[Utente, Depends(get_current_user)],
|
|
db: Session = Depends(get_db),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=100),
|
|
sito_id: Optional[int] = None,
|
|
severita: Optional[str] = None,
|
|
stato: Optional[str] = None,
|
|
):
|
|
"""
|
|
Recupera gli allarmi per il cliente dell'utente autenticato.
|
|
Supporta filtri e paginazione.
|
|
"""
|
|
|
|
# Query base: solo allarmi dei siti del cliente dell'utente
|
|
query = (
|
|
db.query(Allarme)
|
|
.join(Sito)
|
|
.filter(Sito.cliente_id == current_user.cliente_id)
|
|
)
|
|
|
|
# Applica filtri opzionali
|
|
if sito_id:
|
|
query = query.filter(Allarme.sito_id == sito_id)
|
|
if severita:
|
|
query = query.filter(Allarme.severita == severita)
|
|
if stato:
|
|
query = query.filter(Allarme.stato == stato)
|
|
|
|
# Conta totale
|
|
total = query.count()
|
|
|
|
# Applica paginazione e ordinamento
|
|
allarmi = (
|
|
query.order_by(desc(Allarme.timestamp_rilevamento))
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
|
|
return AllarmeList(
|
|
total=total,
|
|
items=allarmi,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.get("/{allarme_id}", response_model=AllarmeResponse)
|
|
async def get_allarme(
|
|
allarme_id: int,
|
|
current_user: Annotated[Utente, Depends(get_current_user)],
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Recupera un singolo allarme per ID"""
|
|
|
|
allarme = (
|
|
db.query(Allarme)
|
|
.join(Sito)
|
|
.filter(
|
|
Allarme.id == allarme_id,
|
|
Sito.cliente_id == current_user.cliente_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not allarme:
|
|
raise HTTPException(status_code=404, detail="Allarme non trovato")
|
|
|
|
return allarme
|
|
|
|
|
|
@router.patch("/{allarme_id}", response_model=AllarmeResponse)
|
|
async def update_allarme(
|
|
allarme_id: int,
|
|
update_data: AllarmeUpdate,
|
|
current_user: Annotated[Utente, Depends(get_current_user)],
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Aggiorna lo stato o le note di un allarme"""
|
|
|
|
allarme = (
|
|
db.query(Allarme)
|
|
.join(Sito)
|
|
.filter(
|
|
Allarme.id == allarme_id,
|
|
Sito.cliente_id == current_user.cliente_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not allarme:
|
|
raise HTTPException(status_code=404, detail="Allarme non trovato")
|
|
|
|
# Aggiorna campi se forniti
|
|
if update_data.stato is not None:
|
|
allarme.stato = update_data.stato
|
|
|
|
# Se viene risolto, registra chi e quando
|
|
if update_data.stato == "risolto":
|
|
from datetime import datetime, timezone
|
|
allarme.risolto_da = f"{current_user.nome} {current_user.cognome}"
|
|
allarme.timestamp_risoluzione = datetime.now(timezone.utc)
|
|
|
|
if update_data.note is not None:
|
|
allarme.note = update_data.note
|
|
|
|
if update_data.risolto_da is not None:
|
|
allarme.risolto_da = update_data.risolto_da
|
|
|
|
db.commit()
|
|
db.refresh(allarme)
|
|
|
|
return allarme
|
|
|
|
|
|
@router.get("/sito/{sito_id}", response_model=AllarmeList)
|
|
async def get_allarmi_by_sito(
|
|
sito_id: int,
|
|
current_user: Annotated[Utente, Depends(get_current_user)],
|
|
db: Session = Depends(get_db),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=100),
|
|
):
|
|
"""Recupera tutti gli allarmi per un sito specifico"""
|
|
|
|
# Verifica che il sito appartenga al cliente dell'utente
|
|
sito = (
|
|
db.query(Sito)
|
|
.filter(
|
|
Sito.id == sito_id,
|
|
Sito.cliente_id == current_user.cliente_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not sito:
|
|
raise HTTPException(status_code=404, detail="Sito non trovato")
|
|
|
|
query = db.query(Allarme).filter(Allarme.sito_id == sito_id)
|
|
|
|
total = query.count()
|
|
|
|
allarmi = (
|
|
query.order_by(desc(Allarme.timestamp_rilevamento))
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
|
|
return AllarmeList(
|
|
total=total,
|
|
items=allarmi,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|