151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, col, func, or_, select
|
|
|
|
from app.database import get_session
|
|
from app.models import EntityCache, EntityFlag
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/entities")
|
|
def list_entities(
|
|
page: int = Query(1, ge=1),
|
|
per_page: int = Query(50, ge=1, le=500),
|
|
domain: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
search: Optional[str] = None,
|
|
available: Optional[bool] = None,
|
|
sort_by: str = Query("entity_id"),
|
|
sort_dir: str = Query("asc", pattern="^(asc|desc)$"),
|
|
favorite: Optional[bool] = None,
|
|
ignored: Optional[bool] = None,
|
|
device_class: Optional[str] = None,
|
|
integration: Optional[str] = None,
|
|
area_id: Optional[str] = None,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
query = select(EntityCache)
|
|
|
|
# Filtres
|
|
if domain:
|
|
domains = [d.strip() for d in domain.split(",")]
|
|
query = query.where(col(EntityCache.domain).in_(domains))
|
|
|
|
if state:
|
|
states = [s.strip() for s in state.split(",")]
|
|
query = query.where(col(EntityCache.state).in_(states))
|
|
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
query = query.where(
|
|
or_(
|
|
col(EntityCache.entity_id).ilike(pattern),
|
|
col(EntityCache.friendly_name).ilike(pattern),
|
|
)
|
|
)
|
|
|
|
if available is not None:
|
|
query = query.where(EntityCache.is_available == available)
|
|
|
|
if device_class:
|
|
query = query.where(EntityCache.device_class == device_class)
|
|
|
|
if integration:
|
|
query = query.where(EntityCache.integration == integration)
|
|
|
|
if area_id:
|
|
query = query.where(EntityCache.area_id == area_id)
|
|
|
|
# Filtres flags (nécessite jointure)
|
|
if favorite is not None or ignored is not None:
|
|
query = query.outerjoin(
|
|
EntityFlag, EntityCache.entity_id == EntityFlag.entity_id
|
|
)
|
|
if favorite is not None:
|
|
query = query.where(EntityFlag.favorite == favorite)
|
|
if ignored is not None:
|
|
query = query.where(EntityFlag.ignored_local == ignored)
|
|
|
|
# Compteur total
|
|
count_query = select(func.count()).select_from(query.subquery())
|
|
total = session.exec(count_query).one()
|
|
|
|
# Tri
|
|
sort_column = getattr(EntityCache, sort_by, EntityCache.entity_id)
|
|
if sort_dir == "desc":
|
|
query = query.order_by(col(sort_column).desc())
|
|
else:
|
|
query = query.order_by(col(sort_column).asc())
|
|
|
|
# Pagination
|
|
offset = (page - 1) * per_page
|
|
query = query.offset(offset).limit(per_page)
|
|
|
|
entities = session.exec(query).all()
|
|
|
|
# Récupérer les flags pour chaque entité
|
|
entity_ids = [e.entity_id for e in entities]
|
|
flags_query = select(EntityFlag).where(col(EntityFlag.entity_id).in_(entity_ids))
|
|
flags = {f.entity_id: f for f in session.exec(flags_query).all()}
|
|
|
|
results = []
|
|
for e in entities:
|
|
d = e.model_dump()
|
|
flag = flags.get(e.entity_id)
|
|
d["favorite"] = flag.favorite if flag else False
|
|
d["ignored_local"] = flag.ignored_local if flag else False
|
|
d["notes"] = flag.notes if flag else ""
|
|
d["original_state"] = flag.original_state if flag else None
|
|
d["disabled_at"] = flag.disabled_at.isoformat() if flag and flag.disabled_at else None
|
|
results.append(d)
|
|
|
|
return {
|
|
"items": results,
|
|
"total": total,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"pages": (total + per_page - 1) // per_page if per_page > 0 else 0,
|
|
}
|
|
|
|
|
|
@router.get("/entities/filters")
|
|
def get_filter_values(session: Session = Depends(get_session)):
|
|
"""Retourne les valeurs disponibles pour les filtres."""
|
|
domains = session.exec(
|
|
select(EntityCache.domain).distinct().order_by(EntityCache.domain)
|
|
).all()
|
|
areas = session.exec(
|
|
select(EntityCache.area_id).where(EntityCache.area_id.is_not(None)).distinct().order_by(EntityCache.area_id) # type: ignore
|
|
).all()
|
|
integrations = session.exec(
|
|
select(EntityCache.integration).where(EntityCache.integration.is_not(None)).distinct().order_by(EntityCache.integration) # type: ignore
|
|
).all()
|
|
device_classes = session.exec(
|
|
select(EntityCache.device_class).where(EntityCache.device_class.is_not(None)).distinct().order_by(EntityCache.device_class) # type: ignore
|
|
).all()
|
|
return {
|
|
"domains": domains,
|
|
"areas": areas,
|
|
"integrations": integrations,
|
|
"device_classes": device_classes,
|
|
}
|
|
|
|
|
|
@router.get("/entities/{entity_id}")
|
|
def get_entity(entity_id: str, session: Session = Depends(get_session)):
|
|
entity = session.get(EntityCache, entity_id)
|
|
if not entity:
|
|
raise HTTPException(status_code=404, detail="Entité non trouvée")
|
|
|
|
d = entity.model_dump()
|
|
flag = session.get(EntityFlag, entity_id)
|
|
d["favorite"] = flag.favorite if flag else False
|
|
d["ignored_local"] = flag.ignored_local if flag else False
|
|
d["notes"] = flag.notes if flag else ""
|
|
d["original_state"] = flag.original_state if flag else None
|
|
d["disabled_at"] = flag.disabled_at.isoformat() if flag and flag.disabled_at else None
|
|
|
|
return d
|