Slackbot
11/29/2022, 8:49 PMJason Lunder
11/29/2022, 8:53 PMfrom __future__ import annotations
from typing import IO, List, Union
import typing as t
from io import BytesIO
from starlette.requests import Request
from multipart.multipart import parse_options_header
from starlette.datastructures import UploadFile
from bentoml.exceptions import BentoMLException
from bentoml._internal.types import FileLike
from <http://bentoml.io|bentoml.io> import File
class WSIFile(File):
def __new__( # pylint: disable=arguments-differ # returning subclass from new
cls, kind = "binaryio", mime_type: str | None = None
) -> File:
mime_type = mime_type if mime_type is not None else "application/octet-stream"
if kind == "binaryio":
res = object.__new__(BytesIOFile)
else:
raise ValueError(f"invalid File kind '{kind}'")
res._mime_type = mime_type
return res
class BytesIOFile(File):
async def from_http_request(self, request: Request) -> IO[bytes]:
# return request
content_type, _ = parse_options_header(request.headers["content-type"])
if content_type.decode("utf-8") == "multipart/form-data":
form = await request.form()
found_mimes: List[str] = []
val: Union[str, UploadFile]
for val in form.values(): # type: ignore
if isinstance(val, UploadFile):
found_mimes.append(val.content_type) # type: ignore (bad starlette types)
if val.content_type == self._mime_type: # type: ignore (bad starlette types)
res = FileLike[bytes](val.file, val.filename) # type: ignore (bad starlette types)
break
if 'image/' in val.content_type:
res = FileLike[bytes](val.file, val.filename) # type: ignore (bad starlette types)
break
else:
if len(found_mimes) == 0:
raise BentoMLException("no File found in multipart form")
else:
raise BentoMLException(
f"multipart File should have Content-Type '{self._mime_type}', got files with content types {', '.join(found_mimes)}"
)
return res # type: ignore
body = await request.body()
return t.cast(IO[bytes], FileLike(BytesIO(body), "<request body>"))Steve T
11/29/2022, 9:34 PMSean
11/29/2022, 10:18 PMto_spec function to all IO descriptors. The fix is for WSIFile to implement the to_spec function. You can reference the to_spec implementations of other IO descriptors. args includes all the keys and values of the arguments used to initialized the IO Descriptor class.
class WSIFile(File, descriptor_id="package.name.WSIFile"):
def to_spec(self) -> dict[str, t.Any]:
return {
"id": self.descriptor_id,
"args": {
...
},
}
This was technically an incompatible change introduced in v1.0.8 so we will fix it in the next release.Steve T
11/29/2022, 10:19 PMSean
11/30/2022, 12:55 AM