161 lines
4.7 KiB
Python
161 lines
4.7 KiB
Python
"""Read-only HTTP routes for the Inventory domain."""
|
|
|
|
from typing import Annotated
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
|
|
from app.identity import require_permission
|
|
from app.inventory.dependencies import get_inventory_service
|
|
from app.inventory.exceptions import (
|
|
DuplicateAssetID,
|
|
InventoryValidationError,
|
|
MinerNotFound,
|
|
)
|
|
from app.inventory.models import (
|
|
InventoryAssetIDHistoryResponse,
|
|
InventoryCreateRequest,
|
|
InventoryListMeta,
|
|
InventoryMetadataUpdateRequest,
|
|
InventoryMinerDetailResponse,
|
|
InventoryMinerListResponse,
|
|
InventorySearchFilters,
|
|
)
|
|
from app.inventory.service import InventoryService
|
|
|
|
|
|
router = APIRouter(prefix="/api/inventory/miners", tags=["inventory"])
|
|
require_inventory_read = require_permission("miners.view")
|
|
require_inventory_write = require_permission("miners.update")
|
|
|
|
InventoryServiceDependency = Annotated[
|
|
InventoryService,
|
|
Depends(get_inventory_service),
|
|
]
|
|
InventoryReadUser = Annotated[dict, Depends(require_inventory_read)]
|
|
InventoryWriteUser = Annotated[dict, Depends(require_inventory_write)]
|
|
InventoryQuery = Annotated[InventorySearchFilters, Query()]
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=InventoryMinerDetailResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_inventory_miner(
|
|
request: InventoryCreateRequest,
|
|
service: InventoryServiceDependency,
|
|
current_user: InventoryWriteUser,
|
|
) -> InventoryMinerDetailResponse:
|
|
del current_user
|
|
try:
|
|
miner = service.create_miner(request)
|
|
except DuplicateAssetID as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
except InventoryValidationError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
return InventoryMinerDetailResponse(data=miner)
|
|
|
|
|
|
@router.patch("/{miner_id}", response_model=InventoryMinerDetailResponse)
|
|
def update_inventory_miner_metadata(
|
|
miner_id: UUID,
|
|
request: InventoryMetadataUpdateRequest,
|
|
service: InventoryServiceDependency,
|
|
current_user: InventoryWriteUser,
|
|
) -> InventoryMinerDetailResponse:
|
|
try:
|
|
miner = service.update_metadata(
|
|
miner_id,
|
|
request,
|
|
actor_user_id=UUID(str(current_user["id"])),
|
|
)
|
|
except MinerNotFound as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Miner not found",
|
|
) from exc
|
|
except DuplicateAssetID as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
except InventoryValidationError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
return InventoryMinerDetailResponse(data=miner)
|
|
|
|
|
|
@router.get("", response_model=InventoryMinerListResponse)
|
|
def list_inventory_miners(
|
|
filters: InventoryQuery,
|
|
service: InventoryServiceDependency,
|
|
current_user: InventoryReadUser,
|
|
) -> InventoryMinerListResponse:
|
|
del current_user
|
|
try:
|
|
miners = service.search_miners(filters)
|
|
except InventoryValidationError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
return InventoryMinerListResponse(
|
|
data=miners,
|
|
meta=InventoryListMeta(
|
|
limit=filters.limit,
|
|
offset=filters.offset,
|
|
returned=len(miners),
|
|
),
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{miner_id}/asset-id-history",
|
|
response_model=InventoryAssetIDHistoryResponse,
|
|
)
|
|
def get_inventory_miner_asset_id_history(
|
|
miner_id: UUID,
|
|
service: InventoryServiceDependency,
|
|
current_user: InventoryReadUser,
|
|
) -> InventoryAssetIDHistoryResponse:
|
|
del current_user
|
|
try:
|
|
history = service.get_asset_id_history(miner_id)
|
|
except MinerNotFound as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Miner not found",
|
|
) from exc
|
|
|
|
return InventoryAssetIDHistoryResponse(data=history)
|
|
|
|
|
|
@router.get("/{miner_id}", response_model=InventoryMinerDetailResponse)
|
|
def get_inventory_miner(
|
|
miner_id: UUID,
|
|
service: InventoryServiceDependency,
|
|
current_user: InventoryReadUser,
|
|
) -> InventoryMinerDetailResponse:
|
|
del current_user
|
|
try:
|
|
miner = service.get_miner(miner_id)
|
|
except MinerNotFound as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Miner not found",
|
|
) from exc
|
|
|
|
return InventoryMinerDetailResponse(data=miner)
|