flask: add common.RegexConverter

based on https://github.com/rhyselsmore/flask-reggie

app.py

@@ -3,9 +3,11 @@ import logging
 
 from flask import Flask
 from flask_caching import Cache
+from oauth_dropins.webutil import appengine_info, appengine_config, handlers, util
 from werkzeug.exceptions import HTTPException
 
-from oauth_dropins.webutil import appengine_info, appengine_config, handlers, util
+import common
+
 
 app = Flask('bridgy-fed')
 app.template_folder = './templates'
@@ -15,6 +17,7 @@ app.config.from_mapping(
     SECRET_KEY=util.read('flask_secret_key'),
     JSONIFY_PRETTYPRINT_REGULAR=True,
 )
+app.url_map.converters['regex'] = common.RegexConverter
 app.wsgi_app = handlers.ndb_context_middleware(
     app.wsgi_app, client=appengine_config.ndb_client)
 

common.py

@@ -14,6 +14,7 @@ from oauth_dropins.webutil import util, webmention
 import requests
 from webob import exc
 from werkzeug.exceptions import abort
+from werkzeug.routing import BaseConverter
 
 from models import Response
 
@@ -87,10 +88,39 @@ DOMAINS = (PRIMARY_DOMAIN,) + OTHER_DOMAINS
 
 
 def not_5xx(resp):
+    """Returns True if resp is an HTTP 5xx, False otherwise.
+
+    Useful to pass to `@cache.cached`'s `response_filter` kwarg to avoid caching
+    5xxes.
+
+    Args:
+      resp: :class:`flask.Response`
+
+    Returns: boolean
+    """
     return (isinstance(resp, tuple) and len(resp) > 1 and util.is_int(resp[1]) and
             resp[1] // 100 != 5)
 
 
+class RegexConverter(BaseConverter):
+    """Regexp URL route for Werkzeug/Flask.
+
+    Based on https://github.com/rhyselsmore/flask-reggie.
+
+    Usage:
+
+      @app.route('/<regex("(abc|def)"):letters>')
+
+    Install with:
+
+      app = Flask(...)
+      app.url_map.converters['regex'] = RegexConverter
+    """
+    def __init__(self, url_map, *items):
+        super(RegexConverter, self).__init__(url_map)
+        self.regex = items[0]
+
+
 def requests_get(url, **kwargs):
     return _requests_fn(util.requests_get, url, **kwargs)
 
@@ -132,10 +162,10 @@ def get_as2(url):
         url: string
 
     Returns:
-        requests.Response
+        :class:`requests.Response`
 
     Raises:
-        requests.HTTPError, webob.exc.HTTPException
+        :class:`requests.HTTPError`, :class:`webob.exc.HTTPException`
 
         If we raise webob HTTPException, it will have an additional response
         attribute with the last requests.Response we received.
@@ -166,7 +196,7 @@ def get_as2(url):
 
 
 def content_type(resp):
-    """Returns a requests.Response's Content-Type, without charset suffix."""
+    """Returns a :class:`requests.Response`'s Content-Type, without charset suffix."""
     type = resp.headers.get('Content-Type')
     if type:
         return type.split(';')[0]
@@ -273,7 +303,8 @@ def postprocess_as2(activity, target=None, key=None):
       activity: dict, AS2 object or activity
       target: dict, AS2 object, optional. The target of activity's inReplyTo or
         Like/Announce/etc object, if any.
-      key: MagicKey, optional. populated into publicKey field if provided.
+      key: :class:`models.MagicKey`, optional. populated into publicKey field
+        if provided.
     """
     type = activity.get('type')
 

tests/test_common.py

@@ -4,7 +4,7 @@ import logging
 import os
 from unittest import mock
 
-from flask import Flask
+from flask import Flask, request
 from oauth_dropins.webutil import util
 from oauth_dropins.webutil.testutil import requests_response
 import requests
@@ -78,6 +78,22 @@ class CommonTest(testutil.TestCase):
                 'inReplyTo': ['foo', 'bar'],
             }))
 
+    def test_regex_converter(self):
+        app = Flask('test_regex_converter')
+        app.url_map.converters['regex'] = common.RegexConverter
+
+        @app.route('/<regex("abc|def"):letters>')
+        def fn(letters):
+            return ''
+
+        with app.test_client() as client:
+            resp = client.get('/def')
+            self.assertEqual(200, resp.status_code)
+            self.assertEqual('def', request.view_args['letters'])
+
+            resp = client.get('/xyz')
+            self.assertEqual(404, resp.status_code)
+
 
 class XrdOrJrdTest(testutil.TestCase):
     def setUp(self):