/
methoddispatcher.py
75 lines (59 loc) · 2.59 KB
/
methoddispatcher.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
""" methoddispatcher
"""
from typing import Dict, Any, Iterable, Callable, List, Tuple
from wsgiref.util import application_uri
from .base import DispatchBase
class MethodDispatcher(DispatchBase):
""" dispatch applications with request method.
"""
def __init__(self, **kwargs) -> None:
super(MethodDispatcher, self).__init__()
for name, app in kwargs.items():
self.register_app(name, app)
def detect_view_name(self, environ: Dict[str, Any]) -> str:
""" convert request method to view name """
return environ['REQUEST_METHOD'].lower()
def on_view_not_found(
self, _,
start_response: Callable[[str, List[Tuple[str, str]]], None],
) -> Iterable[bytes]:
""" called when valid view is not found """
start_response(
"405 Method Not Allowed",
[('Content-type', 'text/plain')])
return [b"Method Not Allowed"]
def action_handler_adapter(handler_cls: type, action_name: str) -> Callable:
""" wraps class to wsgi application dispathing action"""
if not hasattr(handler_cls(), action_name):
message = "{0} does'nt have attr:{1}".format(handler_cls, action_name)
raise ValueError(message)
def wsgiapp(environ, start_response):
""" inner app """
handler = handler_cls()
return getattr(handler, action_name)(environ, start_response)
return wsgiapp
class ActionDispatcher(DispatchBase):
""" wsgi application dispatching actions to registered classes"""
def __init__(self, action_var_name: str = 'action') -> None:
super(ActionDispatcher, self).__init__()
self.action_var_name = action_var_name
def register_actionhandler(self, action_handler: type) -> None:
""" register class as action handler """
for k in action_handler.__dict__:
if k.startswith('_'):
continue
app = action_handler_adapter(action_handler, k)
self.register_app(k, app)
def detect_view_name(self, environ: Dict[str, Any]) -> str:
""" get view name from routing args """
urlvars = environ.get('wsgiorg.routing_args', [(), {}])[1]
return urlvars.get(self.action_var_name)
def on_view_not_found(
self, environ: Dict[str, Any],
start_response: Callable[[str, List[Tuple[str, str]]], None],
) -> Iterable[bytes]:
""" called when action is not found """
start_response(
"404 Not Found",
[('Content-type', 'text/plain')])
return [b"Not Found ", application_uri(environ).encode('utf-8')]