Implementation:Bentoml BentoML IODescriptor Base
| Knowledge Sources | |
|---|---|
| Domains | IO Descriptors, API Specification, HTTP, gRPC, OpenAPI |
| Last Updated | 2026-02-13 15:00 GMT |
Overview
Defines the IODescriptor abstract base class that specifies the input/output data format contract for BentoML inference APIs, including HTTP, gRPC, OpenAPI, and Apache Arrow serialization protocols.
Description
The IODescriptor class is the foundation of BentoML's type-safe IO system. It is a generic abstract base class (IODescriptor[IOType]) that concrete descriptors (Text, JSON, NumpyNdarray, Image, etc.) must subclass. A global IO_DESCRIPTOR_REGISTRY dictionary maps descriptor IDs to their classes, populated automatically via __init_subclass__ when a subclass specifies a descriptor_id.
Key abstract methods that subclasses must implement:
- from_http_request() / to_http_response() for HTTP serialization via Starlette
- from_proto() / to_proto() for gRPC Protocol Buffer serialization
- input_type() for declaring the expected Python input type
- _from_sample() for creating a descriptor from sample data
The class also provides OpenAPI integration via the _OpenAPIMeta mixin, requiring implementations of openapi_schema(), openapi_components(), openapi_example(), openapi_request_body(), and openapi_responses().
Optional methods from_arrow(), to_arrow(), and spark_schema() support batch inference with Apache Arrow and PySpark, raising NotImplementedError by default.
The from_spec() factory function reconstructs an IODescriptor from a dictionary specification, and the from_sample classmethod (auto-generated in __init_subclass__) creates a descriptor instance from sample data.
Usage
This class is not instantiated directly. Subclass it to create new IO descriptor types. Use the concrete subclasses (Text, JSON, NumpyNdarray, etc.) when defining BentoML service API endpoints.
Code Reference
Source Location
- Repository: Bentoml_BentoML
- File: src/bentoml/_internal/io_descriptors/base.py
- Lines: 1-204
Signature
IO_DESCRIPTOR_REGISTRY: dict[str, type[IODescriptor[t.Any]]] = {}
def from_spec(spec: dict[str, t.Any]) -> IODescriptor[t.Any]: ...
class IODescriptor(ABC, _OpenAPIMeta, t.Generic[IOType]):
HTTP_METHODS = ("POST",)
_mime_type: str
descriptor_id: str | None
proto_fields: tuple[ProtoField]
def __init_subclass__(
cls, *, descriptor_id: str | None = None,
proto_fields: tuple[ProtoField] | None = None,
): ...
@property
def sample(self) -> IOType | None: ...
@abstractmethod
def _from_sample(self, sample: t.Any) -> IOType: ...
@property
def mime_type(self) -> str: ...
def to_spec(self) -> dict[str, t.Any] | None: ...
@classmethod
def from_spec(cls, spec: dict[str, t.Any]) -> t.Self: ...
@abstractmethod
def input_type(self) -> InputType: ...
@abstractmethod
async def from_http_request(self, request: Request) -> IOType: ...
@abstractmethod
async def to_http_response(
self, obj: IOType, ctx: Context | None = None
) -> Response | StreamingResponse: ...
@abstractmethod
async def from_proto(self, field: t.Any) -> IOType: ...
@abstractmethod
async def to_proto(self, obj: IOType) -> t.Any: ...
def from_arrow(self, batch: pyarrow.RecordBatch) -> IOType: ...
def to_arrow(self, obj: IOType) -> pyarrow.RecordBatch: ...
def spark_schema(self) -> pyspark.sql.types.StructType: ...
Import
from bentoml._internal.io_descriptors.base import IODescriptor
from bentoml._internal.io_descriptors.base import from_spec
from bentoml._internal.io_descriptors.base import IO_DESCRIPTOR_REGISTRY
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| descriptor_id | str or None | No (subclass kwarg) | Unique identifier for registering the descriptor in the global registry |
| proto_fields | tuple[ProtoField] or None | No (subclass kwarg) | gRPC proto field names for this descriptor |
| request | starlette.requests.Request | Yes (from_http_request) | HTTP request to deserialize |
| obj | IOType | Yes (to_http_response) | Data object to serialize as HTTP response |
| field | Any | Yes (from_proto) | gRPC proto field to deserialize |
| sample | Any | Yes (from_sample) | Sample data for creating a descriptor instance |
| spec | dict[str, Any] | Yes (from_spec) | Specification dictionary with "id" key for registry lookup |
Outputs
| Name | Type | Description |
|---|---|---|
| IOType (from_http_request) | IOType | Deserialized Python object from HTTP request body |
| Response (to_http_response) | Response or StreamingResponse | Serialized HTTP response |
| IOType (from_proto) | IOType | Deserialized Python object from gRPC proto field |
| proto (to_proto) | Any | Serialized gRPC proto message |
| IODescriptor (from_spec) | IODescriptor | Reconstructed descriptor from specification |
Usage Examples
import bentoml
from bentoml.io import Text, JSON, NumpyNdarray
# Using concrete IODescriptor subclasses in a service
svc = bentoml.legacy.Service("my_service", runners=[runner])
@svc.api(input=Text(), output=JSON())
def predict(text: str) -> dict:
return runner.run(text)
# Creating a descriptor from sample data
text_io = Text.from_sample("Hello, world!")
# Reconstructing from spec
from bentoml._internal.io_descriptors.base import from_spec
descriptor = from_spec({"id": "bentoml.io.Text"})
# Implementing a custom IODescriptor
from bentoml._internal.io_descriptors.base import IODescriptor
class CustomDescriptor(
IODescriptor[dict],
descriptor_id="my.custom.Descriptor",
proto_fields=("custom",),
):
_mime_type = "application/json"
def _from_sample(self, sample):
return sample
def input_type(self):
return dict
async def from_http_request(self, request):
return await request.json()
async def to_http_response(self, obj, ctx=None):
from starlette.responses import JSONResponse
return JSONResponse(obj)
async def from_proto(self, field):
raise NotImplementedError
async def to_proto(self, obj):
raise NotImplementedError
def openapi_schema(self):
return {"type": "object"}
def openapi_components(self):
return None
def openapi_example(self):
return self.sample
def openapi_request_body(self):
return {"content": {"application/json": {}}}
def openapi_responses(self):
return {"description": "Success"}