Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #697 from 38elements/stream

Browse Source 
Add Request.stream
This commit is contained in:
Eli Uriegas
2017-05-24 16:22:52 -07:00
committed by GitHub
parent c6d68009d2 a50d8421b8
commit 48de321869
13 changed files with 733 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
sanic/app.py
View File

@@ -60,6 +60,7 @@ class Sanic:
self.sock = None
self.listeners = defaultdict(list)
self.is_running = False
self.is_request_stream = False
self.websocket_enabled = False
self.websocket_tasks = []
@@ -110,12 +111,14 @@ class Sanic:
# Decorator
def route(self, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False):
strict_slashes=False, stream=False):
"""Decorate a function to be registered as a route
:param uri: path of the URL
:param methods: list or tuple of methods allowed
:param host:
:param strict_slashes:
:param stream:
:return: decorated function
"""
@@ -124,7 +127,12 @@ class Sanic:
if not uri.startswith('/'):
uri = '/' + uri
if stream:
self.is_request_stream = True
def response(handler):
if stream:
handler.is_stream = stream
self.router.add(uri=uri, methods=methods, handler=handler,
host=host, strict_slashes=strict_slashes)
return handler
@@ -136,13 +144,13 @@ class Sanic:
return self.route(uri, methods=frozenset({"GET"}), host=host,
strict_slashes=strict_slashes)
def post(self, uri, host=None, strict_slashes=False):
def post(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=frozenset({"POST"}), host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def put(self, uri, host=None, strict_slashes=False):
def put(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=frozenset({"PUT"}), host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def head(self, uri, host=None, strict_slashes=False):
return self.route(uri, methods=frozenset({"HEAD"}), host=host,
@@ -152,9 +160,9 @@ class Sanic:
return self.route(uri, methods=frozenset({"OPTIONS"}), host=host,
strict_slashes=strict_slashes)
def patch(self, uri, host=None, strict_slashes=False):
def patch(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=frozenset({"PATCH"}), host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def delete(self, uri, host=None, strict_slashes=False):
return self.route(uri, methods=frozenset({"DELETE"}), host=host,
@@ -173,20 +181,28 @@ class Sanic:
:param host:
:return: function or class instance
"""
stream = False
# Handle HTTPMethodView differently
if hasattr(handler, 'view_class'):
methods = set()
for method in HTTP_METHODS:
if getattr(handler.view_class, method.lower(), None):
_handler = getattr(handler.view_class, method.lower(), None)
if _handler:
methods.add(method)
if hasattr(_handler, 'is_stream'):
stream = True
# handle composition view differently
if isinstance(handler, CompositionView):
methods = handler.handlers.keys()
for _handler in handler.handlers.values():
if hasattr(_handler, 'is_stream'):
stream = True
break
self.route(uri=uri, methods=methods, host=host,
strict_slashes=strict_slashes)(handler)
strict_slashes=strict_slashes, stream=stream)(handler)
return handler
# Decorator
@@ -664,6 +680,8 @@ class Sanic:
server_settings = {
'protocol': protocol,
'request_class': self.request_class,
'is_request_stream': self.is_request_stream,
'router': self.router,
'host': host,
'port': port,
'sock': sock,

24
sanic/blueprints.py
View File

@@ -5,7 +5,7 @@ from sanic.views import CompositionView
FutureRoute = namedtuple('Route',
['handler', 'uri', 'methods',
'host', 'strict_slashes'])
'host', 'strict_slashes', 'stream'])
FutureListener = namedtuple('Listener', ['handler', 'uri', 'methods', 'host'])
FutureMiddleware = namedtuple('Route', ['middleware', 'args', 'kwargs'])
FutureException = namedtuple('Route', ['handler', 'args', 'kwargs'])
@@ -47,7 +47,8 @@ class Blueprint:
uri=uri[1:] if uri.startswith('//') else uri,
methods=future.methods,
host=future.host or self.host,
strict_slashes=future.strict_slashes
strict_slashes=future.strict_slashes,
stream=future.stream
)(future.handler)
for future in self.websocket_routes:
@@ -88,14 +89,15 @@ class Blueprint:
app.listener(event)(listener)
def route(self, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False):
strict_slashes=False, stream=False):
"""Create a blueprint route from a decorated function.
:param uri: endpoint at which the route will be accessible.
:param methods: list of acceptable HTTP methods.
"""
def decorator(handler):
route = FutureRoute(handler, uri, methods, host, strict_slashes)
route = FutureRoute(
handler, uri, methods, host, strict_slashes, stream)
self.routes.append(route)
return handler
return decorator
@@ -132,7 +134,7 @@ class Blueprint:
:param uri: endpoint at which the route will be accessible.
"""
def decorator(handler):
route = FutureRoute(handler, uri, [], host, strict_slashes)
route = FutureRoute(handler, uri, [], host, strict_slashes, False)
self.websocket_routes.append(route)
return handler
return decorator
@@ -195,13 +197,13 @@ class Blueprint:
return self.route(uri, methods=["GET"], host=host,
strict_slashes=strict_slashes)
def post(self, uri, host=None, strict_slashes=False):
def post(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=["POST"], host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def put(self, uri, host=None, strict_slashes=False):
def put(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=["PUT"], host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def head(self, uri, host=None, strict_slashes=False):
return self.route(uri, methods=["HEAD"], host=host,
@@ -211,9 +213,9 @@ class Blueprint:
return self.route(uri, methods=["OPTIONS"], host=host,
strict_slashes=strict_slashes)
def patch(self, uri, host=None, strict_slashes=False):
def patch(self, uri, host=None, strict_slashes=False, stream=False):
return self.route(uri, methods=["PATCH"], host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, stream=stream)
def delete(self, uri, host=None, strict_slashes=False):
return self.route(uri, methods=["DELETE"], host=host,

3
sanic/request.py
View File

@@ -45,7 +45,7 @@ class Request(dict):
__slots__ = (
'app', 'headers', 'version', 'method', '_cookies', 'transport',
'body', 'parsed_json', 'parsed_args', 'parsed_form', 'parsed_files',
'_ip', '_parsed_url', 'uri_template'
'_ip', '_parsed_url', 'uri_template', 'stream'
)
def __init__(self, url_bytes, headers, version, method, transport):
@@ -66,6 +66,7 @@ class Request(dict):
self.parsed_args = None
self.uri_template = None
self._cookies = None
self.stream = None
@property
def json(self):

11
sanic/router.py
View File

@@ -345,3 +345,14 @@ class Router:
if hasattr(route_handler, 'handlers'):
route_handler = route_handler.handlers[method]
return route_handler, [], kwargs, route.uri
def is_stream_handler(self, request):
""" Handler for request is stream or not.
:param request: Request object
:return: bool
"""
handler = self.get(request)[0]
if (hasattr(handler, 'view_class') and
hasattr(handler.view_class, request.method.lower())):
handler = getattr(handler.view_class, request.method.lower())
return hasattr(handler, 'is_stream')

38
sanic/server.py
View File

@@ -64,22 +64,24 @@ class HttpProtocol(asyncio.Protocol):
'parser', 'request', 'url', 'headers',
# request config
'request_handler', 'request_timeout', 'request_max_size',
'request_class',
'request_class', 'is_request_stream', 'router',
# enable or disable access log / error log purpose
'has_log',
# connection management
'_total_request_size', '_timeout_handler', '_last_communication_time')
'_total_request_size', '_timeout_handler', '_last_communication_time',
'_is_stream_handler')
def __init__(self, *, loop, request_handler, error_handler,
signal=Signal(), connections=set(), request_timeout=60,
request_max_size=None, request_class=None, has_log=True,
keep_alive=True):
keep_alive=True, is_request_stream=False, router=None):
self.loop = loop
self.transport = None
self.request = None
self.parser = None
self.url = None
self.headers = None
self.router = router
self.signal = signal
self.has_log = has_log
self.connections = connections
@@ -88,10 +90,13 @@ class HttpProtocol(asyncio.Protocol):
self.request_timeout = request_timeout
self.request_max_size = request_max_size
self.request_class = request_class or Request
self.is_request_stream = is_request_stream
self._is_stream_handler = False
self._total_request_size = 0
self._timeout_handler = None
self._last_request_time = None
self._request_handler_task = None
self._request_stream_task = None
self._keep_alive = keep_alive
@property
@@ -123,6 +128,8 @@ class HttpProtocol(asyncio.Protocol):
self._timeout_handler = (
self.loop.call_later(time_left, self.connection_timeout))
else:
if self._request_stream_task:
self._request_stream_task.cancel()
if self._request_handler_task:
self._request_handler_task.cancel()
exception = RequestTimeout('Request Timeout')
@@ -171,13 +178,29 @@ class HttpProtocol(asyncio.Protocol):
method=self.parser.get_method().decode(),
transport=self.transport
)
if self.is_request_stream:
self._is_stream_handler = self.router.is_stream_handler(
self.request)
if self._is_stream_handler:
self.request.stream = asyncio.Queue()
self.execute_request_handler()
def on_body(self, body):
if self.is_request_stream and self._is_stream_handler:
self._request_stream_task = self.loop.create_task(
self.request.stream.put(body))
return
self.request.body.append(body)
def on_message_complete(self):
if self.is_request_stream and self._is_stream_handler:
self._request_stream_task = self.loop.create_task(
self.request.stream.put(None))
return
self.request.body = b''.join(self.request.body)
self.execute_request_handler()
def execute_request_handler(self):
self._request_handler_task = self.loop.create_task(
self.request_handler(
self.request,
@@ -319,7 +342,9 @@ class HttpProtocol(asyncio.Protocol):
self.url = None
self.headers = None
self._request_handler_task = None
self._request_stream_task = None
self._total_request_size = 0
self._is_stream_handler = False
def close_if_idle(self):
"""Close the connection if a request is not being sent or received
@@ -361,7 +386,8 @@ def serve(host, port, request_handler, error_handler, before_start=None,
request_timeout=60, ssl=None, sock=None, request_max_size=None,
reuse_port=False, loop=None, protocol=HttpProtocol, backlog=100,
register_sys_signals=True, run_async=False, connections=None,
signal=Signal(), request_class=None, has_log=True, keep_alive=True):
signal=Signal(), request_class=None, has_log=True, keep_alive=True,
is_request_stream=False, router=None):
"""Start asynchronous HTTP Server on an individual process.
:param host: Address to host on
@@ -388,6 +414,8 @@ def serve(host, port, request_handler, error_handler, before_start=None,
:param protocol: subclass of asyncio protocol class
:param request_class: Request class to use
:param has_log: disable/enable access log and error log
:param is_request_stream: disable/enable Request.stream
:param router: Router object
:return: Nothing
"""
if not run_async:
@@ -412,6 +440,8 @@ def serve(host, port, request_handler, error_handler, before_start=None,
request_class=request_class,
has_log=has_log,
keep_alive=keep_alive,
is_request_stream=is_request_stream,
router=router,
)
server_coroutine = loop.create_server(

9
sanic/views.py
View File

@@ -64,6 +64,11 @@ class HTTPMethodView:
return view
def stream(func):
func.is_stream = True
return func
class CompositionView:
"""Simple method-function mapped view for the sanic.
You can add handler functions to methods (get, post, put, patch, delete)
@@ -83,7 +88,9 @@ class CompositionView:
def __init__(self):
self.handlers = {}
def add(self, methods, handler):
def add(self, methods, handler, stream=False):
if stream:
handler.is_stream = stream
for method in methods:
if method not in HTTP_METHODS:
raise InvalidUsage(