PTah
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.jwt_auth import get_current_user
|
|
from app.database import get_db
|
|
from app.models import Event, Host
|
|
from app.schemas.list_models import HostListResponse, HostSummary
|
|
|
|
router = APIRouter(prefix="/hosts", tags=["hosts"])
|
|
|
|
|
|
@router.get("", response_model=HostListResponse)
|
|
def list_hosts(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
product: str | None = None,
|
|
hostname: str | None = None,
|
|
db: Session = Depends(get_db),
|
|
_user: str = Depends(get_current_user),
|
|
) -> HostListResponse:
|
|
stmt = select(Host)
|
|
count_stmt = select(func.count()).select_from(Host)
|
|
|
|
if product:
|
|
stmt = stmt.where(Host.product == product)
|
|
count_stmt = count_stmt.where(Host.product == product)
|
|
if hostname:
|
|
like = f"%{hostname}%"
|
|
stmt = stmt.where(Host.hostname.ilike(like))
|
|
count_stmt = count_stmt.where(Host.hostname.ilike(like))
|
|
|
|
total = db.scalar(count_stmt) or 0
|
|
rows = db.scalars(
|
|
stmt.order_by(Host.last_seen_at.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
).all()
|
|
|
|
items: list[HostSummary] = []
|
|
for host in rows:
|
|
event_count = db.scalar(
|
|
select(func.count()).select_from(Event).where(Event.host_id == host.id)
|
|
)
|
|
items.append(
|
|
HostSummary(
|
|
id=host.id,
|
|
hostname=host.hostname,
|
|
display_name=host.display_name,
|
|
os_family=host.os_family,
|
|
product=host.product,
|
|
product_version=host.product_version,
|
|
ipv4=host.ipv4,
|
|
last_seen_at=host.last_seen_at,
|
|
event_count=int(event_count or 0),
|
|
)
|
|
)
|
|
|
|
return HostListResponse(items=items, total=total, page=page, page_size=page_size)
|