feat(BA-4903): add REST API read endpoints for ResourceSlotType (#9707)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: HyeockJinKim

changes/9707.feature.md

@@ -0,0 +1 @@
+Add REST API read endpoints for ResourceSlotType (`GET /resource-slot-types` with pagination/filter/order, `GET /resource-slot-types/{slot_name}` returning 404 on missing)

src/ai/backend/common/data/permission/types.py

@@ -107,6 +107,7 @@ class EntityType(enum.StrEnum):
     PERMISSION = "permission"
     AGENT_RESOURCE = "agent_resource"
     RESOURCE_ALLOCATION = "resource_allocation"
+    RESOURCE_SLOT_TYPE = "resource_slot_type"
     RESOURCE_GROUP = "resource_group"
     PROMETHEUS_QUERY_PRESET = "prometheus_query_preset"
     RESOURCE_PRESET = "resource_preset"

src/ai/backend/common/dto/manager/resource_slot/__init__.py

@@ -0,0 +1,37 @@
+"""
+Common DTOs for resource slot type management.
+"""
+
+from __future__ import annotations
+
+from .request import (
+    OrderDirection,
+    ResourceSlotTypeFilter,
+    ResourceSlotTypeOrder,
+    ResourceSlotTypeOrderField,
+    ResourceSlotTypePathParam,
+    SearchResourceSlotTypesRequest,
+)
+from .response import (
+    GetResourceSlotTypeResponse,
+    NumberFormatDTO,
+    PaginationInfo,
+    ResourceSlotTypeDTO,
+    SearchResourceSlotTypesResponse,
+)
+
+__all__ = (
+    # Request DTOs
+    "OrderDirection",
+    "ResourceSlotTypeFilter",
+    "ResourceSlotTypeOrder",
+    "ResourceSlotTypeOrderField",
+    "ResourceSlotTypePathParam",
+    "SearchResourceSlotTypesRequest",
+    # Response DTOs
+    "NumberFormatDTO",
+    "ResourceSlotTypeDTO",
+    "PaginationInfo",
+    "SearchResourceSlotTypesResponse",
+    "GetResourceSlotTypeResponse",
+)

src/ai/backend/common/dto/manager/resource_slot/request.py

@@ -0,0 +1,85 @@
+"""
+Request DTOs for Resource Slot Type REST API.
+Shared between Client SDK and Manager API.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+from pydantic import Field
+
+from ai.backend.common.api_handlers import BaseRequestModel
+from ai.backend.common.dto.manager.defs import DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT
+from ai.backend.common.dto.manager.query import StringFilter
+
+__all__ = (
+    "OrderDirection",
+    "ResourceSlotTypeOrderField",
+    "ResourceSlotTypeFilter",
+    "ResourceSlotTypeOrder",
+    "ResourceSlotTypePathParam",
+    "SearchResourceSlotTypesRequest",
+)
+
+
+class OrderDirection(StrEnum):
+    """Order direction for sorting."""
+
+    ASC = "asc"
+    DESC = "desc"
+
+
+class ResourceSlotTypeOrderField(StrEnum):
+    """Fields available for ordering resource slot types."""
+
+    SLOT_NAME = "slot_name"
+    RANK = "rank"
+    DISPLAY_NAME = "display_name"
+
+
+class ResourceSlotTypeFilter(BaseRequestModel):
+    """Filter conditions for resource slot types."""
+
+    slot_name: StringFilter | None = Field(
+        default=None,
+        description=(
+            "Filter by slot name. "
+            "Supports equals, contains, starts_with, ends_with, "
+            "and their case-insensitive and negated variants."
+        ),
+    )
+    slot_type: StringFilter | None = Field(
+        default=None,
+        description="Filter by slot type.",
+    )
+    display_name: StringFilter | None = Field(
+        default=None,
+        description="Filter by display name.",
+    )
+
+
+class ResourceSlotTypeOrder(BaseRequestModel):
+    """Order specification for resource slot types."""
+
+    field: ResourceSlotTypeOrderField = Field(description="Field to order by")
+    direction: OrderDirection = Field(default=OrderDirection.ASC, description="Order direction")
+
+
+class ResourceSlotTypePathParam(BaseRequestModel):
+    """Path parameter for resource slot type slot_name."""
+
+    slot_name: str = Field(description="Resource slot type name")
+
+
+class SearchResourceSlotTypesRequest(BaseRequestModel):
+    """Request body for searching resource slot types with filters, orders, and pagination."""
+
+    filter: ResourceSlotTypeFilter | None = Field(default=None, description="Filter conditions")
+    order: list[ResourceSlotTypeOrder] | None = Field(
+        default=None, description="Order specifications"
+    )
+    limit: int = Field(
+        default=DEFAULT_PAGE_LIMIT, ge=1, le=MAX_PAGE_LIMIT, description="Maximum items to return"
+    )
+    offset: int = Field(default=0, ge=0, description="Number of items to skip")

src/ai/backend/common/dto/manager/resource_slot/response.py

@@ -0,0 +1,59 @@
+"""
+Response DTOs for Resource Slot Type REST API.
+Shared between Client SDK and Manager API.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from ai.backend.common.api_handlers import BaseResponseModel
+
+__all__ = (
+    "NumberFormatDTO",
+    "ResourceSlotTypeDTO",
+    "PaginationInfo",
+    "SearchResourceSlotTypesResponse",
+    "GetResourceSlotTypeResponse",
+)
+
+
+class NumberFormatDTO(BaseModel):
+    """DTO for number format data."""
+
+    binary: bool = Field(description="Whether to use binary (1024-based) units")
+    round_length: int = Field(description="Number of decimal places to round to")
+
+
+class ResourceSlotTypeDTO(BaseModel):
+    """DTO for resource slot type data."""
+
+    slot_name: str = Field(description="Unique slot name identifier")
+    slot_type: str = Field(description="Slot type (e.g., count, bytes)")
+    display_name: str = Field(description="Human-readable display name")
+    description: str = Field(description="Description of the resource slot type")
+    display_unit: str = Field(description="Unit string for display purposes")
+    display_icon: str = Field(description="Icon identifier for display purposes")
+    number_format: NumberFormatDTO = Field(description="Number formatting options")
+    rank: int = Field(description="Display rank/order")
+
+
+class PaginationInfo(BaseModel):
+    """Pagination information."""
+
+    total: int = Field(description="Total number of items")
+    offset: int = Field(description="Number of items skipped")
+    limit: int | None = Field(default=None, description="Maximum items returned")
+
+
+class SearchResourceSlotTypesResponse(BaseResponseModel):
+    """Response for searching resource slot types."""
+
+    items: list[ResourceSlotTypeDTO] = Field(description="List of resource slot types")
+    pagination: PaginationInfo = Field(description="Pagination information")
+
+
+class GetResourceSlotTypeResponse(BaseResponseModel):
+    """Response for getting a single resource slot type."""
+
+    item: ResourceSlotTypeDTO = Field(description="Resource slot type data")

src/ai/backend/manager/api/rest/resource_slot/__init__.py

@@ -0,0 +1,3 @@
+from .registry import register_resource_slot_routes
+
+__all__ = ["register_resource_slot_routes"]

src/ai/backend/manager/api/rest/resource_slot/adapter.py

@@ -0,0 +1,110 @@
+"""
+Adapter to convert ResourceSlotType DTOs to repository Querier objects.
+"""
+
+from __future__ import annotations
+
+from ai.backend.common.dto.manager.resource_slot.request import (
+    OrderDirection,
+    ResourceSlotTypeFilter,
+    ResourceSlotTypeOrder,
+    ResourceSlotTypeOrderField,
+    SearchResourceSlotTypesRequest,
+)
+from ai.backend.common.dto.manager.resource_slot.response import (
+    NumberFormatDTO,
+    ResourceSlotTypeDTO,
+)
+from ai.backend.manager.api.rest.adapter import BaseFilterAdapter
+from ai.backend.manager.data.resource_slot.types import ResourceSlotTypeData
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    OffsetPagination,
+    QueryCondition,
+    QueryOrder,
+)
+from ai.backend.manager.repositories.resource_slot.query import QueryConditions, QueryOrders
+
+__all__ = ("ResourceSlotAdapter",)
+
+
+class ResourceSlotAdapter(BaseFilterAdapter):
+    """Adapter for converting resource slot type requests to repository queries."""
+
+    def convert_to_dto(self, data: ResourceSlotTypeData) -> ResourceSlotTypeDTO:
+        """Convert ResourceSlotTypeData to DTO."""
+        return ResourceSlotTypeDTO(
+            slot_name=data.slot_name,
+            slot_type=data.slot_type,
+            display_name=data.display_name,
+            description=data.description,
+            display_unit=data.display_unit,
+            display_icon=data.display_icon,
+            number_format=NumberFormatDTO(
+                binary=data.number_format.binary,
+                round_length=data.number_format.round_length,
+            ),
+            rank=data.rank,
+        )
+
+    def build_querier(self, request: SearchResourceSlotTypesRequest) -> BatchQuerier:
+        """Build a BatchQuerier from search request."""
+        conditions = self._convert_filter(request.filter) if request.filter else []
+        orders = [self._convert_order(o) for o in request.order] if request.order else []
+
+        return BatchQuerier(
+            conditions=conditions,
+            orders=orders,
+            pagination=OffsetPagination(limit=request.limit, offset=request.offset),
+        )
+
+    def _convert_filter(self, filter: ResourceSlotTypeFilter) -> list[QueryCondition]:
+        """Convert resource slot type filter to list of query conditions."""
+        conditions: list[QueryCondition] = []
+
+        if filter.slot_name is not None:
+            condition = self.convert_string_filter(
+                filter.slot_name,
+                contains_factory=QueryConditions.by_slot_name_contains,
+                equals_factory=QueryConditions.by_slot_name_equals,
+                starts_with_factory=QueryConditions.by_slot_name_starts_with,
+                ends_with_factory=QueryConditions.by_slot_name_ends_with,
+            )
+            if condition is not None:
+                conditions.append(condition)
+
+        if filter.slot_type is not None:
+            condition = self.convert_string_filter(
+                filter.slot_type,
+                contains_factory=QueryConditions.by_slot_type_contains,
+                equals_factory=QueryConditions.by_slot_type_equals,
+                starts_with_factory=QueryConditions.by_slot_type_starts_with,
+                ends_with_factory=QueryConditions.by_slot_type_ends_with,
+            )
+            if condition is not None:
+                conditions.append(condition)
+
+        if filter.display_name is not None:
+            condition = self.convert_string_filter(
+                filter.display_name,
+                contains_factory=QueryConditions.by_display_name_contains,
+                equals_factory=QueryConditions.by_display_name_equals,
+                starts_with_factory=QueryConditions.by_display_name_starts_with,
+                ends_with_factory=QueryConditions.by_display_name_ends_with,
+            )
+            if condition is not None:
+                conditions.append(condition)
+
+        return conditions
+
+    def _convert_order(self, order: ResourceSlotTypeOrder) -> QueryOrder:
+        """Convert resource slot type order specification to query order."""
+        ascending = order.direction == OrderDirection.ASC
+
+        if order.field == ResourceSlotTypeOrderField.SLOT_NAME:
+            return QueryOrders.slot_name(ascending=ascending)
+        if order.field == ResourceSlotTypeOrderField.RANK:
+            return QueryOrders.rank(ascending=ascending)
+        if order.field == ResourceSlotTypeOrderField.DISPLAY_NAME:
+            return QueryOrders.display_name(ascending=ascending)
+        raise ValueError(f"Unknown order field: {order.field}")

src/ai/backend/manager/api/rest/resource_slot/handler.py

@@ -0,0 +1,80 @@
+"""Resource Slot Type handler class using constructor dependency injection."""
+
+from __future__ import annotations
+
+import logging
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Final
+
+from ai.backend.common.api_handlers import APIResponse, BodyParam, PathParam
+from ai.backend.common.dto.manager.resource_slot.request import (
+    ResourceSlotTypePathParam,
+    SearchResourceSlotTypesRequest,
+)
+from ai.backend.common.dto.manager.resource_slot.response import (
+    GetResourceSlotTypeResponse,
+    PaginationInfo,
+    SearchResourceSlotTypesResponse,
+)
+from ai.backend.logging import BraceStyleAdapter
+from ai.backend.manager.services.resource_slot.actions.get_resource_slot_type import (
+    GetResourceSlotTypeAction,
+)
+from ai.backend.manager.services.resource_slot.actions.search_resource_slot_types import (
+    SearchResourceSlotTypesAction,
+)
+
+from .adapter import ResourceSlotAdapter
+
+if TYPE_CHECKING:
+    from ai.backend.manager.services.resource_slot.processors import ResourceSlotProcessors
+
+log: Final = BraceStyleAdapter(logging.getLogger(__spec__.name))
+
+
+class ResourceSlotHandler:
+    """Resource Slot Type API handler with constructor-injected dependencies."""
+
+    def __init__(self, *, resource_slot: ResourceSlotProcessors) -> None:
+        self._resource_slot = resource_slot
+        self._adapter = ResourceSlotAdapter()
+
+    async def search_resource_slot_types(
+        self,
+        body: BodyParam[SearchResourceSlotTypesRequest],
+    ) -> APIResponse:
+        """Search resource slot types with filters, orders, and pagination."""
+        log.info("SEARCH_RESOURCE_SLOT_TYPES")
+
+        querier = self._adapter.build_querier(body.parsed)
+
+        action_result = await self._resource_slot.search_resource_slot_types.wait_for_complete(
+            SearchResourceSlotTypesAction(querier=querier)
+        )
+
+        resp = SearchResourceSlotTypesResponse(
+            items=[self._adapter.convert_to_dto(item) for item in action_result.items],
+            pagination=PaginationInfo(
+                total=action_result.total_count,
+                offset=body.parsed.offset,
+                limit=body.parsed.limit,
+            ),
+        )
+        return APIResponse.build(status_code=HTTPStatus.OK, response_model=resp)
+
+    async def get_resource_slot_type(
+        self,
+        path: PathParam[ResourceSlotTypePathParam],
+    ) -> APIResponse:
+        """Get a single resource slot type by slot_name."""
+        slot_name = path.parsed.slot_name
+        log.info("GET_RESOURCE_SLOT_TYPE (slot_name:{})", slot_name)
+
+        action_result = await self._resource_slot.get_resource_slot_type.wait_for_complete(
+            GetResourceSlotTypeAction(slot_name=slot_name)
+        )
+
+        resp = GetResourceSlotTypeResponse(
+            item=self._adapter.convert_to_dto(action_result.item),
+        )
+        return APIResponse.build(status_code=HTTPStatus.OK, response_model=resp)