add method to create response from a static resource (#12)

rolo/response.py

@@ -1,8 +1,14 @@
 import json
-from typing import Any, Dict, Iterable, Type, Union
+import mimetypes
+import typing as t
+from importlib import resources
 
+from werkzeug.exceptions import NotFound
 from werkzeug.wrappers import Response as WerkzeugResponse
 
+if t.TYPE_CHECKING:
+    from types import ModuleType
+
 
 class Response(WerkzeugResponse):
     """
@@ -22,7 +28,7 @@ def update_from(self, other: WerkzeugResponse):
         self._on_close.extend(other._on_close)
         self.headers.update(other.headers)
 
-    def set_json(self, doc: Any, cls: Type[json.JSONEncoder] = None):
+    def set_json(self, doc: t.Any, cls: t.Type[json.JSONEncoder] = None):
         """
         Serializes the given dictionary using localstack's ``CustomEncoder`` into a json response, and sets the
         mimetype automatically to ``application/json``.
@@ -33,7 +39,7 @@ def set_json(self, doc: Any, cls: Type[json.JSONEncoder] = None):
         self.data = json.dumps(doc, cls=cls)
         self.mimetype = "application/json"
 
-    def set_response(self, response: Union[str, bytes, bytearray, Iterable[bytes]]):
+    def set_response(self, response: t.Union[str, bytes, bytearray, t.Iterable[bytes]]):
         """
         Function to set the low-level ``response`` object. This is copied from the werkzeug Response constructor. The
         response attribute always holds an iterable of bytes. Passing a str, bytes or bytearray is equivalent to
@@ -53,7 +59,7 @@ def set_response(self, response: Union[str, bytes, bytearray, Iterable[bytes]]):
 
         return self
 
-    def to_readonly_response_dict(self) -> Dict:
+    def to_readonly_response_dict(self) -> t.Dict:
         """
         Returns a read-only version of a response dictionary as it is often expected by other libraries like boto.
         """
@@ -64,7 +70,7 @@ def to_readonly_response_dict(self) -> Dict:
         }
 
     @classmethod
-    def for_json(cls, doc: Any, *args, **kwargs) -> "Response":
+    def for_json(cls, doc: t.Any, *args, **kwargs) -> "Response":
         """
         Creates a new JSON response from the given document. It automatically sets the mimetype to ``application/json``.
 
@@ -76,3 +82,22 @@ def for_json(cls, doc: Any, *args, **kwargs) -> "Response":
         response = cls(*args, **kwargs)
         response.set_json(doc)
         return response
+
+    @classmethod
+    def for_resource(cls, module: "ModuleType", path: str, *args, **kwargs) -> "Response":
+        """
+        Looks up the given file in the given module, and creates a new Response object with the contents of that
+        file. It guesses the mimetype of the file and sets it in the response accordingly. If the file does not exist
+        ,it raises a ``NotFound`` error.
+
+        :param module: the module to look up the file in
+        :param path: the path/file name
+        :return: a new Response object
+        """
+        resource = resources.files(module).joinpath(path)
+        if not resource.is_file():
+            raise NotFound()
+        mimetype = mimetypes.guess_type(resource.name)
+        mimetype = mimetype[0] if mimetype and mimetype[0] else "application/octet-stream"
+
+        return cls(resource.open("rb"), *args, mimetype=mimetype, **kwargs)

rolo/router.py

@@ -24,6 +24,9 @@
 
 from .request import get_raw_path
 
+if t.TYPE_CHECKING:
+    from _typeshed.wsgi import WSGIApplication
+
 HTTP_METHODS = ("GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS", "TRACE")
 
 E = TypeVar("E")
@@ -403,6 +406,23 @@ def wrapper(fn):
 
         return wrapper
 
+    def wsgi(self) -> "WSGIApplication":
+        """
+        Returns this router as a WSGI compatible interface. This can be used to conveniently serve a Router instance
+        through a WSGI server, for instance werkzeug's dev server::
+
+            from werkzeug.serving import run_simple
+
+            from rolo import Router
+            from rolo.dispatcher import handler_dispatcher
+
+            router = Router(dispatcher=handler_dispatcher())
+            run_simple("localhost", 5000, router.wsgi())
+
+        :return: a WSGI callable that invokes this router
+        """
+        return Request.application(self.dispatch)
+
 
 class RuleAdapter(RuleFactory):
     """

tests/static/index.html

@@ -0,0 +1,3 @@
+<html lang="en">
+<body>hello</body>
+</html>

tests/static/test.txt

@@ -0,0 +1 @@
+hello world

tests/test_response.py

@@ -0,0 +1,84 @@
+import io
+
+import pytest
+from werkzeug.exceptions import NotFound
+
+from rolo import Response
+from tests import static
+
+
+def test_for_resource_html():
+    response = Response.for_resource(static, "index.html")
+    assert response.content_type == "text/html; charset=utf-8"
+    assert response.get_data() == b'<html lang="en">\n<body>hello</body>\n</html>\n'
+    assert response.status == "200 OK"
+
+
+def test_for_resource_txt():
+    response = Response.for_resource(static, "test.txt")
+    assert response.content_type == "text/plain; charset=utf-8"
+    assert response.get_data() == b"hello world\n"
+    assert response.status == "200 OK"
+
+
+def test_for_resource_with_custom_response_status_and_headers():
+    response = Response.for_resource(static, "test.txt", status=201, headers={"X-Foo": "Bar"})
+    assert response.content_type == "text/plain; charset=utf-8"
+    assert response.get_data() == b"hello world\n"
+    assert response.status == "201 CREATED"
+    assert response.headers.get("X-Foo") == "Bar"
+
+
+def test_for_resource_not_found():
+    with pytest.raises(NotFound):
+        Response.for_resource(static, "doesntexist.txt")
+
+
+def test_for_json():
+    response = Response.for_json(
+        {"foo": "bar", "420": 69, "isTrue": True},
+    )
+    assert response.content_type == "application/json"
+    assert response.get_data() == b'{"foo": "bar", "420": 69, "isTrue": true}'
+    assert response.status == "200 OK"
+
+
+def test_for_json_with_custom_response_status_and_headers():
+    response = Response.for_json(
+        {"foo": "bar", "420": 69, "isTrue": True},
+        status=201,
+        headers={"X-Foo": "Bar"},
+    )
+    assert response.content_type == "application/json"
+    assert response.get_data() == b'{"foo": "bar", "420": 69, "isTrue": true}'
+    assert response.status == "201 CREATED"
+    assert response.headers.get("X-Foo") == "Bar"
+
+
+@pytest.mark.parametrize(
+    argnames="data",
+    argvalues=[
+        b"foobar",
+        "foobar",
+        io.BytesIO(b"foobar"),
+        [b"foo", b"bar"],
+    ],
+)
+def test_set_response(data):
+    response = Response()
+    response.set_response(data)
+    assert response.get_data() == b"foobar"
+
+
+def test_update_from():
+    original = Response(
+        [b"foo", b"bar"], 202, headers={"X-Foo": "Bar"}, mimetype="application/octet-stream"
+    )
+
+    response = Response()
+    response.update_from(original)
+
+    assert response.get_data() == b"foobar"
+    assert response.status_code == 202
+    assert response.headers.get("X-Foo") == "Bar"
+    assert response.content_type == "application/octet-stream"