67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
from fastapi import APIRouter, Depends, Query, UploadFile, File, Form
|
|
from fastapi.responses import FileResponse
|
|
from typing import Optional
|
|
|
|
from auth.models import TokenPayload
|
|
from auth.dependencies import require_permission
|
|
from firmware.models import FirmwareVersion, FirmwareListResponse, FirmwareLatestResponse
|
|
from firmware import service
|
|
|
|
router = APIRouter(prefix="/api/firmware", tags=["firmware"])
|
|
|
|
|
|
@router.post("/upload", response_model=FirmwareVersion, status_code=201)
|
|
async def upload_firmware(
|
|
hw_type: str = Form(...),
|
|
channel: str = Form(...),
|
|
version: str = Form(...),
|
|
notes: Optional[str] = Form(None),
|
|
file: UploadFile = File(...),
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "add")),
|
|
):
|
|
file_bytes = await file.read()
|
|
return service.upload_firmware(
|
|
hw_type=hw_type,
|
|
channel=channel,
|
|
version=version,
|
|
file_bytes=file_bytes,
|
|
notes=notes,
|
|
)
|
|
|
|
|
|
@router.get("", response_model=FirmwareListResponse)
|
|
def list_firmware(
|
|
hw_type: Optional[str] = Query(None),
|
|
channel: Optional[str] = Query(None),
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "view")),
|
|
):
|
|
items = service.list_firmware(hw_type=hw_type, channel=channel)
|
|
return FirmwareListResponse(firmware=items, total=len(items))
|
|
|
|
|
|
@router.get("/{hw_type}/{channel}/latest", response_model=FirmwareLatestResponse)
|
|
def get_latest_firmware(hw_type: str, channel: str):
|
|
"""Returns metadata for the latest firmware for a given hw_type + channel.
|
|
No auth required — devices call this endpoint to check for updates.
|
|
"""
|
|
return service.get_latest(hw_type, channel)
|
|
|
|
|
|
@router.get("/{hw_type}/{channel}/{version}/firmware.bin")
|
|
def download_firmware(hw_type: str, channel: str, version: str):
|
|
"""Download the firmware binary. No auth required — devices call this directly."""
|
|
path = service.get_firmware_path(hw_type, channel, version)
|
|
return FileResponse(
|
|
path=str(path),
|
|
media_type="application/octet-stream",
|
|
filename="firmware.bin",
|
|
)
|
|
|
|
|
|
@router.delete("/{firmware_id}", status_code=204)
|
|
def delete_firmware(
|
|
firmware_id: str,
|
|
_user: TokenPayload = Depends(require_permission("manufacturing", "delete")),
|
|
):
|
|
service.delete_firmware(firmware_id)
|