Added FileResponseModel and StreamingResponseModel for file and streaming response.

Author: eadwinCode

ellar/core/response/model/__init__.py

@@ -6,7 +6,12 @@
     RouteResponseExecution,
 )
 from .factory import create_response_model
-from .html import HTMLResponseModel
+from .file import (
+    FileResponseModel,
+    StreamingResponseModel,
+    StreamingResponseModelInvalidContent,
+)
+from .html import HTMLResponseModel, HTMLResponseModelRuntimeError
 from .interface import IResponseModel
 from .json import EmptyAPIResponseModel, JSONResponseModel
 from .route import RouteResponseModel
@@ -18,9 +23,13 @@
     "ResponseModelField",
     "JSONResponseModel",
     "EmptyAPIResponseModel",
+    "FileResponseModel",
+    "StreamingResponseModel",
     "RouteResponseModel",
     "ResponseTypeDefinitionConverter",
     "IResponseModel",
     "HTMLResponseModel",
     "create_response_model",
+    "HTMLResponseModelRuntimeError",
+    "StreamingResponseModelInvalidContent",
 ]

ellar/core/response/model/base.py

@@ -67,17 +67,17 @@ class BaseResponseModel(IResponseModel, ABC):
     __slots__ = (
         "_response_type",
         "_media_type",
-        "_description",
         "meta",
         "_model_field",
     )
 
     response_type: t.Type[Response] = Response
     model_field_or_schema: t.Union[ResponseModelField, t.Any] = None
+    default_description: str = "Successful Response"
 
     def __init__(
         self,
-        description: str = "Successful Response",
+        description: str = None,
         model_field_or_schema: t.Union[ResponseModelField, t.Any] = None,
         **kwargs: t.Any,
     ) -> None:
@@ -87,7 +87,7 @@ def __init__(
         self._media_type = str(
             kwargs.get("media_type") or self._response_type.media_type
         )
-        self._description = description
+        self.default_description = description or self.default_description
         self.meta = kwargs
         self._model_field = self._get_model_field_from_schema(model_field_or_schema)
 
@@ -97,7 +97,7 @@ def media_type(self) -> str:
 
     @property
     def description(self) -> str:
-        return self._description
+        return self.default_description
 
     def _get_model_field_from_schema(
         self, model_field_or_schema: t.Optional[t.Union[ResponseModelField, t.Any]]
@@ -139,7 +139,9 @@ def create_response(
         self, context: IExecutionContext, response_obj: t.Any, status_code: int
     ) -> Response:
         """Cant create custom responses, Please override this function to create a custom response"""
-        response_args, headers = self.get_context_response(context=context)
+        response_args, headers = self.get_context_response(
+            context=context, status_code=status_code
+        )
         serializer_filter: t.Optional[SerializerFilter] = reflect.get_metadata(
             SERIALIZER_FILTER_KEY, context.get_handler()
         )

ellar/core/response/model/factory.py

@@ -12,7 +12,7 @@ def create_response_model(
     response_model: t.Union[t.Type["ResponseModel"], t.Type[T]],
     *,
     response_type: t.Type[Response] = None,
-    description: str = "Successful Response",
+    description: str = None,
     model_field_or_schema: t.Union[t.Type["ResponseModelField"], t.Any] = None,
     **kwargs: t.Any,
 ) -> t.Union["ResponseModel", T]:

ellar/core/response/model/file.py

@@ -0,0 +1,96 @@
+import typing as t
+from enum import Enum
+
+from ellar.core.context import IExecutionContext
+from ellar.serializer import Serializer, SerializerFilter, serialize_object
+
+from ..response_types import FileResponse, Response, StreamingResponse
+from .base import ResponseModel, ResponseModelField
+
+
+class StreamingResponseModelInvalidContent(RuntimeError):
+    pass
+
+
+class ContentDispositionType(str, Enum):
+    inline = "inline"
+    attachment = "attachment"
+
+
+class FileResponseModelSchema(Serializer):
+    path: str
+    media_type: t.Optional[str] = None
+    filename: t.Optional[str] = None
+    method: t.Optional[str] = None
+    content_disposition_type: ContentDispositionType = ContentDispositionType.attachment
+
+
+class FileResponseModel(ResponseModel):
+    __slots__ = ("_file_init_schema",)
+
+    response_type: t.Type[Response] = FileResponse
+    file_init_schema_type = FileResponseModelSchema
+
+    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
+        super(FileResponseModel, self).__init__(*args, **kwargs)
+        self._file_init_schema = self._get_model_field_from_schema(
+            self.file_init_schema_type
+        )
+
+    def create_response(
+        self, context: IExecutionContext, response_obj: t.Any, status_code: int
+    ) -> Response:
+        response_args, headers = self.get_context_response(
+            context=context, status_code=status_code
+        )
+
+        init_kwargs = serialize_object(self.serialize(response_obj))
+        response_args.update(init_kwargs)
+
+        response = self._response_type(
+            **response_args,
+            headers=headers,
+        )
+        return response
+
+    def get_init_kwargs_schema(self) -> ResponseModelField:
+        assert self._file_init_schema
+        return self._file_init_schema
+
+    def get_model_field(self) -> t.Optional[t.Union[ResponseModelField, t.Any]]:
+        # We don't want any schema for this.
+        return None
+
+    def serialize(
+        self,
+        response_obj: t.Any,
+        serializer_filter: t.Optional[SerializerFilter] = None,
+    ) -> t.Union[t.List[t.Dict], t.Dict, t.Any]:
+        _response_model_field = self.get_init_kwargs_schema()
+        return _response_model_field.serialize(
+            response_obj, serializer_filter=serializer_filter
+        )
+
+
+class StreamingResponseModel(ResponseModel):
+    response_type = StreamingResponse
+
+    def get_model_field(self) -> t.Optional[t.Union[ResponseModelField, t.Any]]:
+        # We don't want any schema for this.
+        return None
+
+    def create_response(
+        self, context: IExecutionContext, response_obj: t.Any, status_code: int
+    ) -> Response:
+        response_args, headers = self.get_context_response(
+            context=context, status_code=status_code
+        )
+        if not isinstance(response_obj, (t.AsyncGenerator, t.Generator)):
+            raise StreamingResponseModelInvalidContent(
+                "Content must typing.AsyncIterable OR typing.Iterable"
+            )
+
+        response = self._response_type(
+            **response_args, headers=headers, content=response_obj
+        )
+        return response