-
Notifications
You must be signed in to change notification settings - Fork 756
/
base.py
123 lines (94 loc) · 3.16 KB
/
base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from __future__ import annotations
import typing as t
from abc import ABCMeta
from abc import abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from types import UnionType
from typing_extensions import Self
from starlette.requests import Request
from starlette.responses import Response
from bentoml.grpc.v1.service_pb2 import Request as GRPCRequest
from bentoml.grpc.v1.service_pb2 import Response as GRPCResponse
from ..types import LazyType
from ..context import InferenceApiContext as Context
from ..server.grpc.types import BentoServicerContext
InputType = (
UnionType
| t.Type[t.Any]
| LazyType[t.Any]
| dict[str, t.Type[t.Any] | UnionType | LazyType[t.Any]]
)
IOPyObj = t.TypeVar("IOPyObj")
_T = t.TypeVar("_T")
class DescriptorMeta(ABCMeta):
def __new__(
cls: type[Self],
name: str,
bases: tuple[type, ...],
namespace: dict[str, t.Any],
*,
proto_fields: list[str] | None = None,
) -> Self:
if not proto_fields:
proto_fields = []
namespace["_proto_fields"] = proto_fields
return super().__new__(cls, name, bases, namespace)
class IODescriptor(t.Generic[IOPyObj], metaclass=DescriptorMeta, proto_fields=None):
"""
IODescriptor describes the input/output data format of an InferenceAPI defined
in a :code:`bentoml.Service`. This is an abstract base class for extending new HTTP
endpoint IO descriptor types in BentoServer.
"""
HTTP_METHODS = ["POST"]
_init_str: str = ""
_proto_fields: list[str]
def __new__(cls: t.Type[_T], *args: t.Any, **kwargs: t.Any) -> _T:
self = super().__new__(cls)
arg_strs = tuple(repr(i) for i in args) + tuple(
f"{k}={repr(v)}" for k, v in kwargs.items()
)
setattr(self, "_init_str", f"{cls.__name__}({', '.join(arg_strs)})")
return self
def __repr__(self) -> str:
return self._init_str
@property
def accepted_proto_fields(self) -> list[str]:
"""
Returns a list of kinds fields that the IODescriptor can accept.
Make sure to keep in sync with bentoml.grpc.v1.Value message.
"""
return self._proto_fields
@abstractmethod
def input_type(self) -> InputType:
...
@abstractmethod
def openapi_schema_type(self) -> dict[str, str]:
...
@abstractmethod
def openapi_request_schema(self) -> dict[str, t.Any]:
...
@abstractmethod
def openapi_responses_schema(self) -> dict[str, t.Any]:
...
@abstractmethod
async def from_http_request(self, request: Request) -> IOPyObj:
...
@abstractmethod
async def to_http_response(
self, obj: IOPyObj, ctx: Context | None = None
) -> Response:
...
@abstractmethod
def generate_protobuf(self):
...
@abstractmethod
async def from_grpc_request(
self, request: GRPCRequest, context: BentoServicerContext
) -> IOPyObj:
...
@abstractmethod
async def to_grpc_response(
self, obj: IOPyObj, context: BentoServicerContext
) -> GRPCResponse:
...