120 lines
3.0 KiB
Python
120 lines
3.0 KiB
Python
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api import deps
|
|
from app.db.session import get_db
|
|
from app.models.device import Device
|
|
from app.models.user import User
|
|
from app.schemas.device import DeviceCreate, DeviceUpdate, DeviceResponse
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[DeviceResponse])
|
|
def get_devices(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_user),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
search: Optional[str] = None,
|
|
is_active: Optional[bool] = None,
|
|
) -> Any:
|
|
"""
|
|
Retrieve devices with optional filtering.
|
|
"""
|
|
query = db.query(Device)
|
|
|
|
if search:
|
|
query = query.filter(Device.name.ilike(f"%{search}%") | Device.id.ilike(f"%{search}%"))
|
|
|
|
if is_active is not None:
|
|
query = query.filter(Device.is_active == is_active)
|
|
|
|
devices = query.offset(skip).limit(limit).all()
|
|
return devices
|
|
|
|
|
|
@router.post("/", response_model=DeviceResponse)
|
|
def create_device(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_user),
|
|
device_in: DeviceCreate,
|
|
) -> Any:
|
|
"""
|
|
Create new device.
|
|
"""
|
|
device = Device(
|
|
id=device_in.id,
|
|
name=device_in.name,
|
|
description=device_in.description,
|
|
model=device_in.model,
|
|
firmware_version=device_in.firmware_version,
|
|
)
|
|
db.add(device)
|
|
db.commit()
|
|
db.refresh(device)
|
|
return device
|
|
|
|
|
|
@router.get("/{device_id}", response_model=DeviceResponse)
|
|
def get_device(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_user),
|
|
device_id: str,
|
|
) -> Any:
|
|
"""
|
|
Get device by ID.
|
|
"""
|
|
device = db.query(Device).filter(Device.id == device_id).first()
|
|
if not device:
|
|
raise HTTPException(status_code=404, detail="Device not found")
|
|
return device
|
|
|
|
|
|
@router.put("/{device_id}", response_model=DeviceResponse)
|
|
def update_device(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_user),
|
|
device_id: str,
|
|
device_in: DeviceUpdate,
|
|
) -> Any:
|
|
"""
|
|
Update device.
|
|
"""
|
|
device = db.query(Device).filter(Device.id == device_id).first()
|
|
if not device:
|
|
raise HTTPException(status_code=404, detail="Device not found")
|
|
|
|
update_data = device_in.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(device, field, value)
|
|
|
|
db.add(device)
|
|
db.commit()
|
|
db.refresh(device)
|
|
return device
|
|
|
|
|
|
@router.delete("/{device_id}")
|
|
def delete_device(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(deps.get_current_user),
|
|
device_id: str,
|
|
) -> Any:
|
|
"""
|
|
Delete device.
|
|
"""
|
|
device = db.query(Device).filter(Device.id == device_id).first()
|
|
if not device:
|
|
raise HTTPException(status_code=404, detail="Device not found")
|
|
|
|
db.delete(device)
|
|
db.commit()
|
|
return {"success": True}
|