WIP: New style routing (#47)

* Remove staticfile and server adapter

* Implement new style routing

* Correspond cast error

* Fix flake8 and mypy

* Update example

example/app.py

@@ -1,16 +1,17 @@
 from kobin import Kobin, request, response, template
+from wsgi_static_middleware import StaticMiddleware
 
 app = Kobin()
 app.config.load_from_pyfile('config.py')
 
 
-@app.route('^/$')
+@app.route('/')
 def index():
-    response.add_header("hoge", "fuga")
+    response.headers.add_header("hoge", "fuga")
     return template('hello_jinja2', name='Kobin')
 
 
-@app.route('^/user/(?P<name>\w+)$')
+@app.route('/user/{name}')
 def hello(name: str):
     return """
     <p>Hello {}</p>
@@ -19,4 +20,8 @@ def hello(name: str):
     """.format(name, request.path, str(response.headerlist))
 
 if __name__ == '__main__':
-    app.run()
+    from wsgiref.simple_server import make_server
+    app = StaticMiddleware(app, static_root=app.config['STATIC_URL'],
+                           static_dirs=app.config['STATICFILES_DIRS'])
+    httpd = make_server('', 8080, app)
+    httpd.serve_forever()

example/config.py

@@ -3,8 +3,4 @@
 BASE_DIR = os.path.dirname(os.path.abspath(__name__))
 TEMPLATE_DIRS = [os.path.join(BASE_DIR, 'templates')]
 STATICFILES_DIRS = [os.path.join(BASE_DIR, 'static')]
-STATIC_URL = '/static/'
-
-PORT = 8080
-HOST = '127.0.0.1'
-SERVER = 'wsgiref'
+STATIC_URL = 'static'

kobin/__init__.py

@@ -3,4 +3,3 @@
 from .templates import template, jinja2_template
 from .exceptions import HTTPError
 from .routes import redirect
-

kobin/app.py

@@ -10,32 +10,11 @@ class Kobin:
     def __init__(self, root_path: str='.') -> None:
         self.router = Router()
         self.config = Config(os.path.abspath(root_path))
-        # register static files view
-        from .static_files import static_file
-        route = Route('^{}(?P<filename>.*)'.format(self.config['STATIC_ROOT']), 'GET', static_file)
-        self.add_route(route)
 
-    def run(self, server: str='wsgiref', **kwargs) -> None:
-        from .server_adapters import ServerAdapter, servers
-        try:
-            if server not in servers:
-                raise ImportError('{server} is not supported.'.format(server))
-            server_cls = servers.get(self.config['SERVER'])
-            server_obj = server_cls(host=self.config['HOST'],
-                                    port=self.config['PORT'], **kwargs)  # type: ServerAdapter
-            print('Serving on port {}...'.format(self.config['PORT']))
-            server_obj.run(self)
-        except KeyboardInterrupt:
-            print('Goodbye.')
-
-    def add_route(self, route: Route) -> None:
-        self.router.add(route.rule, route.method, route)
-
-    def route(self, path: str=None, method: str='GET',
+    def route(self, rule: str=None, method: str='GET', name: str=None,
               callback: Callable[..., Union[str, bytes]]=None) -> Callable[..., Union[str, bytes]]:
         def decorator(callback_func):
-            route = Route(path, method, callback_func)
-            self.add_route(route)
+            self.router.add(method, rule, name, callback_func)
             return callback_func
         return decorator(callback) if callback else decorator
 
@@ -44,8 +23,8 @@ def _handle(self, environ: Dict) -> Union[str, bytes]:
         request.bind(environ)  # type: ignore
         response.bind()        # type: ignore
         try:
-            route, kwargs = self.router.match(environ)
-            output = route.call(**kwargs) if kwargs else route.call()
+            callback, kwargs = self.router.match(environ)
+            output = callback(**kwargs) if kwargs else callback()
         except HTTPError:
             import sys
             _type, _value, _traceback = sys.exc_info()

kobin/environs.py

@@ -93,7 +93,7 @@ def url(self) -> str:
 
     @property
     def cookies(self) -> Dict[str, str]:
-        cookies = SimpleCookie(self.environ.get('HTTP_COOKIE', '')).values()
+        cookies = SimpleCookie(self.environ.get('HTTP_COOKIE', '')).values()  # type: ignore
         return {c.key: c.value for c in cookies}
 
     def get_cookie(self, key: str, default: str=None, secret=None) -> str:
@@ -156,7 +156,7 @@ def __init__(self, body: str='', status: int=None, headers: Dict=None,
         self.headers = Headers()
         self.body = body
         self._status_code = status or self.default_status
-        self._cookies = SimpleCookie()  # type: SimpleCookie
+        self._cookies = SimpleCookie()  # type: ignore
 
         if headers:
             for name, value in headers.items():
@@ -217,7 +217,7 @@ def set_cookie(self, key: str, value: Any, expires: str=None, path: str=None, **
                 elif isinstance(v, (int, float)):
                     v = v.gmtime(value)
                 v = time.strftime("%a, %d %b %Y %H:%M:%S GMT", v)  # type: ignore
-            self._cookies[key][k.replace('_', '-')] = v
+            self._cookies[key][k.replace('_', '-')] = v  # type: ignore
 
     def delete_cookie(self, key, **kwargs) -> None:
         kwargs['max_age'] = -1

kobin/routes.py

@@ -1,4 +1,3 @@
-import re
 from urllib.parse import urljoin
 from typing import Callable, Dict, List, Tuple, Union, Any, get_type_hints  # type: ignore
 
@@ -9,60 +8,95 @@
 DEFAULT_ARG_TYPE = str
 
 
-def type_args(args_dict: Dict[str, str], type_hints: Dict[str, Any]) -> Dict[str, Any]:
-    for k, v in args_dict.items():
-        arg_type = type_hints.get(k, DEFAULT_ARG_TYPE)
-        args_dict[k] = arg_type(v)
-        return args_dict
-
-
 def redirect(url):
     status = 303 if request.get('SERVER_PROTOCOL') == "HTTP/1.1" else 302
     response.status = status
-    response.add_header('Location', urljoin(request.url, url))
+    response.headers.add_header('Location', urljoin(request.url, url))
     return ""
 
 
+def split_by_slash(path: str) -> List[str]:
+    stripped_path = path.lstrip('/').rstrip('/')
+    return stripped_path.split('/')
+
+
 class Route:
     """ This class wraps a route callback along with route specific metadata.
         It is also responsible for turing an URL path rule into a regular
         expression usable by the Router.
     """
-    def __init__(self, rule: str, method: str, callback: Callable[..., Union[str, bytes]]) -> None:
+    def __init__(self, rule: str, method: str, name: str,
+                 callback: Union[str, bytes]) -> None:
         self.rule = rule
-        self.method = method
+        self.method = method.upper()
+        self.name = name
         self.callback = callback
-        self.callback_types = get_type_hints(callback)  # type: Dict[str, Any]
 
-    def call(self, **kwargs) -> Union[str, bytes]:
-        return self.callback(**kwargs)
+    @property
+    def callback_types(self) -> Dict[str, Any]:
+        return get_type_hints(self.callback)  # type: ignore
+
+    def get_typed_url_vars(self, url_vars: Dict[str, str]) -> Dict[str, Any]:
+        typed_url_vars = {}  # type: Dict[str, Any]
+        for k, v in url_vars.items():
+            arg_type = self.callback_types.get(k, DEFAULT_ARG_TYPE)
+            typed_url_vars[k] = arg_type(v)
+        return typed_url_vars
+
+    def _match_method(self, method: str) -> bool:
+        return self.method == method.upper()
+
+    def _match_path(self, path: str) -> Union[None, Dict[str, Any]]:
+        split_rule = split_by_slash(self.rule)
+        split_path = split_by_slash(path)
+        url_vars = {}  # type: Dict[str, str]
+
+        if len(split_rule) != len(split_path):
+            return  # type: ignore
+
+        for r, p in zip(split_rule, split_path):
+            if r.startswith('{') and r.endswith('}'):
+                url_var_key = r.lstrip('{').rstrip('}')
+                url_vars[url_var_key] = p
+                continue
+            if r != p:
+                return  # type: ignore
+        try:
+            typed_url_vars = self.get_typed_url_vars(url_vars)
+        except ValueError:
+            return  # type: ignore
+        return typed_url_vars
+
+    def match(self, method: str, path: str) -> Dict[str, Any]:
+        if not self._match_method(method):
+            return None
+
+        url_vars = self._match_path(path)
+        if url_vars is not None:
+            return self.get_typed_url_vars(url_vars)
 
 
 class Router:
     def __init__(self) -> None:
-        # Search structure for static route
-        self.routes = {}  # type: Dict[str, Dict[str, List[Any]]]
+        self.routes = []  # type: List['Route']
 
-    def match(self, environ: Dict[str, str]) -> Tuple[Route, Dict[str, Any]]:
+    def match(self, environ: Dict[str, str]) \
+            -> Tuple[Callable[..., Union[str, bytes]], Dict[str, Any]]:
         method = environ['REQUEST_METHOD'].upper()
         path = environ['PATH_INFO'] or '/'
 
-        if method not in self.routes:
-            raise HTTPError(405, "Method Not Allowed: {}".format(method))
-
-        for p in self.routes[method]:
-            pattern = re.compile(p)
-            if pattern.search(path):
-                route, getargs = self.routes[method][p]
-                return route, getargs(path)
-        else:
-            raise HTTPError(404, "Not found: {}".format(repr(path)))
+        for route in self.routes:
+            url_vars = route.match(method, path)
+            if url_vars is not None:
+                return route.callback, url_vars  # type: ignore
+        raise HTTPError(status=404, body='Not found: {}'.format(request.path))
 
-    def add(self, rule: str, method: str, route: Route) -> None:
+    def add(self, method: str, rule: str, name: str, callback: Union[str, bytes]) -> None:
         """ Add a new rule or replace the target for an existing rule. """
-        def getargs(path: str) -> Dict[str, Any]:
-            args_dict = re.compile(rule).match(path).groupdict()
-            return type_args(args_dict, route.callback_types)
+        route = Route(method=method.upper(), rule=rule, name=name, callback=callback)
+        self.routes.append(route)
 
-        self.routes.setdefault(method, {})
-        self.routes[method][rule] = (route, getargs)  # type: ignore
+    def reverse(self, name, **kwargs) -> str:
+        for route in self.routes:
+            if name == route.name:
+                return route.rule.format(**kwargs)

setup.cfg

@@ -3,7 +3,7 @@ release = register sdist bdist_wheel upload
 
 [flake8]
 max-line-length = 120
-exclude = venv/*.py,build/*.py,*/__init__.py,doc/*.py
+exclude = venv/*.py,build/*.py,*/__init__.py
 ignore = F401
 
 [pytest]