Add abort function

src/microdot.py

@@ -594,6 +594,15 @@ class URLPattern():
         return args
 
 
+class HTTPException(Exception):
+    def __init__(self, status_code, reason=None):
+        self.status_code = status_code
+        self.reason = reason or str(status_code) + ' error'
+
+    def __repr__(self):  # pragma: no cover
+        return 'HTTPException: {}'.format(self.status_code)
+
+
 class Microdot():
     """An HTTP application class.
 
@@ -816,6 +825,28 @@ class Microdot():
         for status_code, handler in subapp.error_handlers.items():
             self.error_handlers[status_code] = handler
 
+    @staticmethod
+    def abort(status_code, reason=None):
+        """Abort the current request and return an error response with the
+        given status code.
+
+        :param status_code: The numeric status code of the response.
+        :param reason: The reason for the response, which is included in the
+                       response body.
+
+        Example::
+
+            from microdot import abort
+
+            @app.route('/users/<int:id>')
+            def get_user(id):
+                user = get_user_by_id(id)
+                if user is None:
+                    abort(404)
+                return user.to_dict()
+        """
+        raise HTTPException(status_code, reason)
+
     def run(self, host='0.0.0.0', port=5000, debug=False):
         """Start the web server. This function does not normally return, as
         the server enters an endless listening loop. The :func:`shutdown`
@@ -962,6 +993,12 @@ class Microdot():
                         res = self.error_handlers[f](req)
                     else:
                         res = 'Not found', f
+                except HTTPException as exc:
+                    print_exception(exc)
+                    if exc.status_code in self.error_handlers:
+                        res = self.error_handlers[exc.status_code](req)
+                    else:
+                        res = exc.reason, exc.status_code
                 except Exception as exc:
                     print_exception(exc)
                     res = None
@@ -988,5 +1025,6 @@ class Microdot():
         return res
 
 
+abort = Microdot.abort
 redirect = Response.redirect
 send_file = Response.send_file

src/microdot_asyncio.py

@@ -20,6 +20,7 @@ from microdot import Microdot as BaseMicrodot
 from microdot import print_exception
 from microdot import Request as BaseRequest
 from microdot import Response as BaseResponse
+from microdot import HTTPException
 
 
 def _iscoroutine(coro):
@@ -360,6 +361,12 @@ class Microdot(BaseMicrodot):
                             self.error_handlers[f], req)
                     else:
                         res = 'Not found', f
+                except HTTPException as exc:
+                    print_exception(exc)
+                    if exc.status_code in self.error_handlers:
+                        res = self.error_handlers[exc.status_code](req)
+                    else:
+                        res = exc.reason, exc.status_code
                 except Exception as exc:
                     print_exception(exc)
                     res = None
@@ -393,5 +400,6 @@ class Microdot(BaseMicrodot):
         return ret
 
 
+abort = Microdot.abort
 redirect = Response.redirect
 send_file = Response.send_file

tests/test_microdot.py

@@ -1,6 +1,6 @@
 import sys
 import unittest
-from microdot import Microdot, Response
+from microdot import Microdot, Response, abort
 from microdot_test_client import TestClient
 from tests import mock_socket
 
@@ -439,6 +439,38 @@ class TestMicrodot(unittest.TestCase):
         self.assertEqual(res.headers['Content-Type'], 'text/plain')
         self.assertEqual(res.text, '501')
 
+    def test_abort(self):
+        app = Microdot()
+
+        @app.route('/')
+        def index(req):
+            abort(406, 'Not acceptable')
+            return 'foo'
+
+        client = TestClient(app)
+        res = client.get('/')
+        self.assertEqual(res.status_code, 406)
+        self.assertEqual(res.headers['Content-Type'], 'text/plain')
+        self.assertEqual(res.text, 'Not acceptable')
+
+    def test_abort_handler(self):
+        app = Microdot()
+
+        @app.route('/')
+        def index(req):
+            abort(406)
+            return 'foo'
+
+        @app.errorhandler(406)
+        def handle_406(req):
+            return '406', 406
+
+        client = TestClient(app)
+        res = client.get('/')
+        self.assertEqual(res.status_code, 406)
+        self.assertEqual(res.headers['Content-Type'], 'text/plain')
+        self.assertEqual(res.text, '406')
+
     def test_json_response(self):
         app = Microdot()
 

tests/test_microdot_asyncio.py

@@ -4,7 +4,7 @@ except ImportError:
     import asyncio
 import sys
 import unittest
-from microdot_asyncio import Microdot, Response
+from microdot_asyncio import Microdot, Response, abort
 from microdot_asyncio_test_client import TestClient
 from tests import mock_asyncio, mock_socket
 
@@ -472,6 +472,38 @@ class TestMicrodotAsync(unittest.TestCase):
         self.assertEqual(res.headers['Content-Type'], 'text/plain')
         self.assertEqual(res.text, '501')
 
+    def test_abort(self):
+        app = Microdot()
+
+        @app.route('/')
+        def index(req):
+            abort(406, 'Not acceptable')
+            return 'foo'
+
+        client = TestClient(app)
+        res = self._run(client.get('/'))
+        self.assertEqual(res.status_code, 406)
+        self.assertEqual(res.headers['Content-Type'], 'text/plain')
+        self.assertEqual(res.text, 'Not acceptable')
+
+    def test_abort_handler(self):
+        app = Microdot()
+
+        @app.route('/')
+        def index(req):
+            abort(406)
+            return 'foo'
+
+        @app.errorhandler(406)
+        def handle_500(req):
+            return '406', 406
+
+        client = TestClient(app)
+        res = self._run(client.get('/'))
+        self.assertEqual(res.status_code, 406)
+        self.assertEqual(res.headers['Content-Type'], 'text/plain')
+        self.assertEqual(res.text, '406')
+
     def test_json_response(self):
         app = Microdot()