reverse_url

frontik/app.py

@@ -5,11 +5,11 @@
 import os
 import re
 import time
-from lxml import etree
 
 import tornado.autoreload
 import tornado.ioloop
 import tornado.web
+from lxml import etree
 from tornado.concurrent import Future
 from tornado.httpclient import AsyncHTTPClient
 from tornado.options import options
@@ -20,6 +20,7 @@
 from frontik.compat import iteritems
 from frontik.handler import ErrorHandler
 from frontik.loggers.request import RequestLogger
+from frontik.util import reverse_regex
 
 app_logger = logging.getLogger('frontik.app')
 
@@ -131,9 +132,22 @@ def handle_404(self, application, request, logger, **kwargs):
 
 
 class RegexpDispatcher(object):
-    def __init__(self, app_list, name='RegexpDispatcher'):
+    def __init__(self, handlers, name='RegexpDispatcher'):
         self.name = name
-        self.handlers = [(re.compile(pattern), handler) for pattern, handler in app_list]
+        self.handlers = []
+        self.handler_names = {}
+
+        for handler_spec in handlers:
+            if len(handler_spec) > 2:
+                pattern, handler, handler_name = handler_spec
+            else:
+                handler_name = None
+                pattern, handler = handler_spec
+
+            self.handlers.append((re.compile(pattern), handler))
+
+            if handler_name is not None:
+                self.handler_names[handler_name] = pattern
 
         app_logger.info('initialized %r', self)
 
@@ -154,14 +168,20 @@ def __call__(self, application, request, logger, **kwargs):
         logger.error('match for request url "%s" not found', request.uri)
         return ErrorHandler(application, request, logger, status_code=404, **kwargs)
 
+    def reverse(self, name, *args, **kwargs):
+        if name not in self.handler_names:
+            raise KeyError('%s not found in named urls' % name)
+
+        return reverse_regex(self.handler_names[name], *args, **kwargs)
+
     def __repr__(self):
         return '{}.{}(<{} routes>)'.format(__package__, self.__class__.__name__, len(self.handlers))
 
 
 def app_dispatcher(tornado_app, request, **kwargs):
     request_id = request.headers.get('X-Request-Id', FrontikApplication.next_request_id())
     request_logger = RequestLogger(request, request_id)
-    return tornado_app.dispatcher(tornado_app, request, request_logger, **kwargs)
+    return tornado_app._dispatcher(tornado_app, request, request_logger, **kwargs)
 
 
 class FrontikApplication(tornado.web.Application):
@@ -171,13 +191,12 @@ class DefaultConfig(object):
         pass
 
     def __init__(self, **settings):
-        tornado_settings = settings.get('tornado_settings')
+        self.start_time = time.time()
 
+        tornado_settings = settings.get('tornado_settings')
         if tornado_settings is None:
             tornado_settings = {}
 
-        self.start_time = time.time()
-
         super(FrontikApplication, self).__init__([
             (r'/version/?', VersionHandler),
             (r'/status/?', StatusHandler),
@@ -194,9 +213,12 @@ def __init__(self, **settings):
         AsyncHTTPClient.configure('tornado.curl_httpclient.CurlAsyncHTTPClient', max_clients=options.max_http_clients)
         self.http_client = self.curl_http_client = AsyncHTTPClient()
 
-        self.dispatcher = RegexpDispatcher(self.application_urls(), self.app)
+        self._dispatcher = RegexpDispatcher(self.application_urls(), self.app)
         self.loggers_initializers = frontik.loggers.bootstrap_app_loggers(self)
 
+    def reverse_url(self, name, *args, **kwargs):
+        return self._dispatcher.reverse(name, *args, **kwargs)
+
     def application_urls(self):
         return [
             ('', FileMappingDispatcher(importlib.import_module('{}.pages'.format(self.app))))

frontik/handler.py

@@ -136,6 +136,9 @@ def redirect(self, url, *args, **kwargs):
         self.log.info('redirecting to: %s', url)
         return super(BaseHandler, self).redirect(url, *args, **kwargs)
 
+    def reverse_url(self, name, *args, **kwargs):
+        return self.application.reverse_url(name, *args, **kwargs)
+
     @staticmethod
     def add_callback(callback):
         IOLoop.current().add_callback(callback)

frontik/util.py

@@ -260,3 +260,29 @@ def asciify_url(url):
 
 def get_cookie_or_url_param_value(handler, param_name):
     return handler.get_argument(param_name, handler.get_cookie(param_name, None))
+
+
+def reverse_regex(pattern, *args, **kwargs):
+    class GroupReplacer(object):
+        def __init__(self, args, kwargs):
+            self.args, self.kwargs = args, kwargs
+            self.current_arg = 0
+
+        def __call__(self, match):
+            value = ''
+            named_group = re.search(r'^\?P<(\w+)>(.*?)$', match.group(1))
+
+            if named_group:
+                group_name = named_group.group(1)
+                if group_name in self.kwargs:
+                    value = self.kwargs[group_name]
+                elif self.current_arg < len(self.args):
+                    value = self.args[self.current_arg]
+                    self.current_arg += 1
+                else:
+                    raise ValueError('Cannot reverse url regex: required number of arguments not found')
+
+            return any_to_unicode(value)
+
+    result = re.sub(r'\(([^)]+)\)', GroupReplacer(args, kwargs), to_unicode(pattern))
+    return result.replace('^', '').replace('$', '')

tests/projects/re_app/config.py

@@ -18,6 +18,7 @@
 
 urls = [
     ('/id/(?P<id>[^/]+)', pages.id_param.Page),
+    ('/id/(?P<id1>[^/]+)/(?P<id2>[^/]+)', pages.handler_404.Page, 'two_ids'),
     ('/not_simple', pages.simple.Page),
     ('/exception_on_prepare_regex', pages.exception_on_prepare.Page),
     ('(?!/not_matching_regex)', FileMappingDispatcher(pages, handler_404=pages.handler_404.Page))

tests/projects/re_app/pages/reverse_url.py

@@ -0,0 +1,18 @@
+# coding=utf-8
+
+from frontik.handler import PageHandler
+
+
+class Page(PageHandler):
+    def get_page(self):
+        if self.get_argument('fail_args', 'false') != 'false':
+            self.text = self.reverse_url('two_ids', 1)
+
+        if self.get_argument('fail_missing', 'false') != 'false':
+            self.text = self.reverse_url('missing', 1)
+
+        self.json.put({
+            'args': self.reverse_url('two_ids', 1, 2),
+            'args_and_kwargs': self.reverse_url('two_ids', 2, id1=1),
+            'kwargs': self.reverse_url('two_ids', id1=1, id2=2),
+        })

tests/test_routing.py

@@ -60,3 +60,18 @@ def test_filemapping_custom_404(self):
         response = frontik_re_app.get_page('inexistent_page')
         self.assertEqual(response.status_code, 404)
         self.assertEqual(response.content, b'404')
+
+    def test_reverse_url(self):
+        json = frontik_re_app.get_page_json('reverse_url')
+        self.assertEqual(json, {
+            'args': '/id/1/2',
+            'args_and_kwargs': '/id/1/2',
+            'kwargs': '/id/1/2',
+        })
+
+    def test_reverse_url_fail(self):
+        response = frontik_re_app.get_page('reverse_url?fail_args=true')
+        self.assertEqual(response.status_code, 500)
+
+        response = frontik_re_app.get_page('reverse_url?fail_missing=true')
+        self.assertEqual(response.status_code, 500)

tests/test_util.py

@@ -6,7 +6,7 @@
 from tornado.escape import to_unicode
 from tornado.httputil import HTTPFile, parse_body_arguments
 
-from frontik.util import any_to_bytes, any_to_unicode, make_mfd, make_qs, make_url
+from frontik.util import any_to_bytes, any_to_unicode, make_mfd, make_qs, make_url, reverse_regex
 
 
 class TestUtil(unittest.TestCase):
@@ -115,3 +115,21 @@ def test_make_mfd(self):
         self.assertEqual(files['file3'][1]['filename'], r'file3-\part2\.unknown')
         self.assertEqual(files['file3'][1]['body'], b'BODY2')
         self.assertEqual(files['file3'][1]['content_type'], 'application/octet-stream')
+
+    def test_reverse_regex(self):
+        two_ids = '/id/(?P<id1>[^/]+)/(?P<id2>[^/]+)'
+        two_ids_with_ending = '/id/(?P<id1>[^/]+)/(?P<id2>[^/]+)(\?|$)'
+        two_ids_with_unnamed_groups = '/id/(?P<id1>[^/]+)/(\w+)/(?P<id2>[^/]+)(\?|$)'
+
+        self.assertEqual('/id/1/2', reverse_regex(two_ids, 1, 2))
+        self.assertEqual('/id/1/2', reverse_regex(two_ids_with_ending, 1, 2))
+        self.assertEqual('/id/1//2', reverse_regex(two_ids_with_unnamed_groups, 1, 2))
+        self.assertEqual('/id/1/2', reverse_regex(two_ids, 1, id2=2, id3=3))
+        self.assertEqual('/id/1/2', reverse_regex(two_ids_with_ending, 2, 3, id1='1'))
+        self.assertEqual('/id/1//2', reverse_regex(two_ids_with_unnamed_groups, '1', id2=2))
+        self.assertEqual('/id/1/2', reverse_regex(two_ids, id1=1, id2=2))
+        self.assertEqual('/id/1/2', reverse_regex(two_ids_with_ending, id1='1', id2=2))
+        self.assertEqual('/id/1//2', reverse_regex(two_ids_with_unnamed_groups, id1=1, id2='2'))
+
+        self.assertRaises(ValueError, reverse_regex, two_ids, 1)
+        self.assertRaises(ValueError, reverse_regex, two_ids, id1=1)