""" Linux BenchTools - Devices API """ import json from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.orm import Session from typing import List from app.db.session import get_db from app.schemas.device import DeviceListResponse, DeviceDetail, DeviceSummary, DeviceUpdate from app.schemas.benchmark import BenchmarkSummary from app.schemas.hardware import HardwareSnapshotResponse from app.models.device import Device from app.models.benchmark import Benchmark from app.models.hardware_snapshot import HardwareSnapshot router = APIRouter() @router.get("/devices", response_model=DeviceListResponse) async def get_devices( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), search: str = Query(None), db: Session = Depends(get_db) ): """ Get paginated list of devices with their last benchmark """ query = db.query(Device) # Apply search filter if search: search_filter = f"%{search}%" query = query.filter( (Device.hostname.like(search_filter)) | (Device.description.like(search_filter)) | (Device.tags.like(search_filter)) | (Device.location.like(search_filter)) ) # Get total count total = query.count() # Apply pagination offset = (page - 1) * page_size devices = query.offset(offset).limit(page_size).all() # Build response with last benchmark for each device items = [] for device in devices: # Get last benchmark last_bench = db.query(Benchmark).filter( Benchmark.device_id == device.id ).order_by(Benchmark.run_at.desc()).first() last_bench_summary = None if last_bench: last_bench_summary = BenchmarkSummary( id=last_bench.id, run_at=last_bench.run_at.isoformat(), global_score=last_bench.global_score, cpu_score=last_bench.cpu_score, memory_score=last_bench.memory_score, disk_score=last_bench.disk_score, network_score=last_bench.network_score, gpu_score=last_bench.gpu_score, bench_script_version=last_bench.bench_script_version ) items.append(DeviceSummary( id=device.id, hostname=device.hostname, fqdn=device.fqdn, description=device.description, asset_tag=device.asset_tag, location=device.location, owner=device.owner, tags=device.tags, created_at=device.created_at.isoformat(), updated_at=device.updated_at.isoformat(), last_benchmark=last_bench_summary )) return DeviceListResponse( items=items, total=total, page=page, page_size=page_size ) @router.get("/devices/{device_id}", response_model=DeviceDetail) async def get_device( device_id: int, db: Session = Depends(get_db) ): """ Get detailed information about a specific device """ device = db.query(Device).filter(Device.id == device_id).first() if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Device {device_id} not found" ) # Get last benchmark last_bench = db.query(Benchmark).filter( Benchmark.device_id == device.id ).order_by(Benchmark.run_at.desc()).first() last_bench_summary = None if last_bench: last_bench_summary = BenchmarkSummary( id=last_bench.id, run_at=last_bench.run_at.isoformat(), global_score=last_bench.global_score, cpu_score=last_bench.cpu_score, memory_score=last_bench.memory_score, disk_score=last_bench.disk_score, network_score=last_bench.network_score, gpu_score=last_bench.gpu_score, bench_script_version=last_bench.bench_script_version ) # Get last hardware snapshot last_snapshot = db.query(HardwareSnapshot).filter( HardwareSnapshot.device_id == device.id ).order_by(HardwareSnapshot.captured_at.desc()).first() last_snapshot_data = None if last_snapshot: last_snapshot_data = HardwareSnapshotResponse( id=last_snapshot.id, device_id=last_snapshot.device_id, captured_at=last_snapshot.captured_at.isoformat(), cpu_vendor=last_snapshot.cpu_vendor, cpu_model=last_snapshot.cpu_model, cpu_cores=last_snapshot.cpu_cores, cpu_threads=last_snapshot.cpu_threads, cpu_base_freq_ghz=last_snapshot.cpu_base_freq_ghz, cpu_max_freq_ghz=last_snapshot.cpu_max_freq_ghz, ram_total_mb=last_snapshot.ram_total_mb, ram_slots_total=last_snapshot.ram_slots_total, ram_slots_used=last_snapshot.ram_slots_used, gpu_summary=last_snapshot.gpu_summary, gpu_model=last_snapshot.gpu_model, storage_summary=last_snapshot.storage_summary, storage_devices_json=last_snapshot.storage_devices_json, network_interfaces_json=last_snapshot.network_interfaces_json, os_name=last_snapshot.os_name, os_version=last_snapshot.os_version, kernel_version=last_snapshot.kernel_version, architecture=last_snapshot.architecture, virtualization_type=last_snapshot.virtualization_type, motherboard_vendor=last_snapshot.motherboard_vendor, motherboard_model=last_snapshot.motherboard_model ) return DeviceDetail( id=device.id, hostname=device.hostname, fqdn=device.fqdn, description=device.description, asset_tag=device.asset_tag, location=device.location, owner=device.owner, tags=device.tags, created_at=device.created_at.isoformat(), updated_at=device.updated_at.isoformat(), last_benchmark=last_bench_summary, last_hardware_snapshot=last_snapshot_data ) @router.get("/devices/{device_id}/benchmarks") async def get_device_benchmarks( device_id: int, limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), db: Session = Depends(get_db) ): """ Get benchmark history for a device """ device = db.query(Device).filter(Device.id == device_id).first() if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Device {device_id} not found" ) # Get benchmarks benchmarks = db.query(Benchmark).filter( Benchmark.device_id == device_id ).order_by(Benchmark.run_at.desc()).offset(offset).limit(limit).all() total = db.query(Benchmark).filter(Benchmark.device_id == device_id).count() items = [ BenchmarkSummary( id=b.id, run_at=b.run_at.isoformat(), global_score=b.global_score, cpu_score=b.cpu_score, memory_score=b.memory_score, disk_score=b.disk_score, network_score=b.network_score, gpu_score=b.gpu_score, bench_script_version=b.bench_script_version ) for b in benchmarks ] return { "items": items, "total": total, "limit": limit, "offset": offset } @router.put("/devices/{device_id}", response_model=DeviceDetail) async def update_device( device_id: int, update_data: DeviceUpdate, db: Session = Depends(get_db) ): """ Update device information """ device = db.query(Device).filter(Device.id == device_id).first() if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Device {device_id} not found" ) # Update only provided fields update_dict = update_data.dict(exclude_unset=True) for key, value in update_dict.items(): setattr(device, key, value) device.updated_at = db.query(Device).filter(Device.id == device_id).first().updated_at db.commit() db.refresh(device) # Return updated device (reuse get_device logic) return await get_device(device_id, db)