utp_service/backend/app/routers/global_devices.py
2026-04-07 23:46:54 +03:00

58 lines
1.9 KiB
Python

"""Global devices endpoint — cross-object device listing with filters."""
from typing import Optional
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.dependencies import get_current_user, get_db
from app.models.device import Device
from app.models.object import Object
from app.models.user import User
from app.schemas.device import DeviceRead
router = APIRouter(prefix="/api/v1/devices", tags=["devices-global"])
@router.get("", response_model=list[DeviceRead])
async def list_all_devices(
category: Optional[list[str]] = None,
city: Optional[str] = None,
client: Optional[str] = None,
object_id: Optional[int] = None,
status: Optional[str] = None,
is_active: Optional[bool] = None,
db: AsyncSession = Depends(get_db),
_: User = Depends(get_current_user),
):
"""List devices across all objects with optional filters.
Typical use cases:
- All main_server devices: ?category=main_server
- All bank terminals for a client: ?category=bank_terminal&client=Авангард
- All devices in a city: ?city=Санкт-Петербург
"""
query = (
select(Device)
.join(Object, Device.object_id == Object.id)
.options(selectinload(Device.rack))
.order_by(Object.city, Object.name, Device.category, Device.ip)
)
if category:
query = query.where(Device.category.in_(category))
if city is not None:
query = query.where(Object.city == city)
if client is not None:
query = query.where(Object.client == client)
if object_id is not None:
query = query.where(Device.object_id == object_id)
if status is not None:
query = query.where(Device.status == status)
if is_active is not None:
query = query.where(Device.is_active == is_active)
result = await db.execute(query)
return result.scalars().all()