Skip to content
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
245 lines (206 sloc) 7.88 KB
"""Grail is an async/await-native web framework.
Grail is based on curio, hyper-h2, h11 and other modern
libraries and techniques.
import collections
import functools
import json
import re
import urllib.parse
from curio.kernel import run
from curio.local import Local
from import tcp_server
from curio.socket import *
import h11
from http11 import Http11Connection
from reasons import REASONS
_ctx = Local()
request = Local()
log = Local()
def methodsingledispatch(func):
dispatcher = functools.singledispatch(func)
def wrapper(*args, **kwds):
return dispatcher.dispatch(args[1].__class__)(*args, **kwds)
wrapper.register = dispatcher.register
functools.update_wrapper(wrapper, func)
return wrapper
class Grail:
def __init__(self, import_name):
self.import_name = import_name
self.rules = []
def route(self, query, **kwds):
def adder(f):
self.add_url_rule(query, f.__name__, f, **kwds)
return adder
def add_url_rule(self, target, name, handler, *, methods=['GET']):
self.rules.append(Route(target, name, handler))
def find_rule(self, query):
for route in self.rules:
name, handler, params = route.match(query)
if params is not None:
return Handler(name, handler, Params(params))
raise RoutingError(f'No such route: {query}')
def run_forever(self):
run(tcp_server('', 8080, self.handle_client))
except Exception as exc:
except KeyboardInterrupt as exc:
# Clear the "^C" and extra line
print('\r \033[F')
async def handle_client(self, client, addr):
# this lets us access the app from free functions = self
conn = Http11Connection(client, addr)
server = HttpServer(self, conn)
Handler = collections.namedtuple('Handler', 'name handler params')
class Params:
def __init__(self, d):
self.d = d
def __getattr__(self, name):
return self.d[name]
def get(self, name, *, default=None, type=None):
item = self.d.get(name, default)
if type is None:
return item
if item is None:
return type()
return type(item)
class Route:
def __init__(self, template, name, handler): = name
self.handler = handler
self.trailing_slash = template.endswith('/')
pattern = re.sub(r'\{([a-z]+)\}', '(?P<\\1>[^/]+)', template)
if self.trailing_slash:
pattern += '?'
self.pattern = re.compile(f'^{pattern}$')
print(f'route {name}: {template} => {self.pattern}')
def match(self, query):
m = self.pattern.match(query)
if m is not None:
if self.trailing_slash and not query.endswith('/'):
return (, self.redirector, {})
return (, self.handler, m.groupdict())
return (None, None, None)
async def redirector(self):
return redirect(url_for(, code=308)
def url_for(name, **kwds):
for route in
if == name:
query_params = []
target = route.pattern.pattern[1:-2] # strip ^...?$
for k, v in kwds.items():
target, n = re.subn(f'(?P<{k}>[^)]*)', v, target)
if n == 0:
query_params.append((k, v))
if query_params:
target += '?' + '&'.join(f'{k}={v}' for k, v in query_params)
return target
raise RoutingError(f'no such route: {name}')
def redirect(url, code=302):
assert 300 <= code <= 399
return Response(code, headers={'location': url})
def abort(code, reason=None):
assert 400 <= code <= 599
raise ResponseException(Response(code, reason=reason))
class Response:
def __init__(self, status_code, *, headers=None, data=None, reason=None):
self.status_code = status_code
self.headers = headers or {}
self.reason = reason or REASONS[status_code] = data
if is None and 'content-length' not in self.headers:
self.headers['content-length'] = b'0'
class HttpServer:
def __init__(self, app, conn): = app
self.conn = conn = []
async def send_response(self, res):
headers = collections.ChainMap({}, res.headers)
headers['connection'] = 'close'
await self.conn.send(h11.Response(
async def send_data(self, data):
await self.conn.send(h11.Data(data=data))
async def dispatch(self, query):
name, handler, request.params =
except RoutingError as exc:
await self.send_response(Response(404))
print(f'handling {query} using {name}')
await self.run_handler(handler)
await self.conn.send(h11.EndOfMessage())
await self.conn.send(h11.ConnectionClosed())
async def run_handler(self, handler):
res = await handler()
except ArgumentError as exc:
await self.send_response(Response(400))
except ResponseException as exc:
await self.send_response(exc.res)
await self.handle_response(res)
async def handle_response(self, res):
raise NotImplementedError(f'response: {res!r}, type: {type(res)}')
async def handle_response_Response(self, res):
await self.send_response(res)
if is not None:
await self.send_data('utf-8'))
async def handle_response_bytes(self, res):
await self.send_response(Response(200,
headers={'content-length': str(len(res)).encode('utf-8')}))
await self.send_data(res)
async def handle_response_str(self, res):
data = res.encode('utf-8')
await self.send_response(Response(200,
headers={'content-length': str(len(data)).encode('utf-8')}))
await self.send_data(data)
async def run(self):
e = await self.conn.next_event()
while not isinstance(e, h11.ConnectionClosed):
await self.handle_event(e)
e = await self.conn.next_event()
async def handle_event(self, ev):
raise NotImplementedError(type(ev))
async def handle_event_Request(self, ev):
self.method = ev.method
self.headers = dict(ev.headers) =
async def handle_event_Data(self, ev):
async def handle_event_EndOfMessage(self, ev):
request.method = self.method
request.headers = self.headers
data = b''.join(
if 'content-type' in self.headers:
content_type = self.headers['content-type']
if content_type == 'application/json':
self.form = Params(json.loads(data))
elif content_type == 'application/x-www-form-urlencoded':
self.form = Params(urllib.parse.parse_qs(data))
raise ArgumentError('Unsupported form data Content-type: {content_type}')
await self.dispatch('ascii'))
class GrailError(Exception): pass
class RoutingError(GrailError): pass
class ProtocolError(GrailError): pass
class ArgumentError(GrailError, AttributeError): pass
class ResponseException(BaseException):
def __init__(self, res):
self.res = res
You can’t perform that action at this time.