440 lines
16 KiB
Python
440 lines
16 KiB
Python
"""
|
|
microdot_asyncio
|
|
----------------
|
|
|
|
The ``microdot_asyncio`` module defines a few classes that help implement
|
|
HTTP-based servers for MicroPython and standard Python that use ``asyncio``
|
|
and coroutines.
|
|
"""
|
|
try:
|
|
import uasyncio as asyncio
|
|
except ImportError:
|
|
import asyncio
|
|
|
|
try:
|
|
import uio as io
|
|
except ImportError:
|
|
import io
|
|
|
|
from microdot import Microdot as BaseMicrodot
|
|
from microdot import mro
|
|
from microdot import NoCaseDict
|
|
from microdot import Request as BaseRequest
|
|
from microdot import Response as BaseResponse
|
|
from microdot import print_exception
|
|
from microdot import HTTPException
|
|
from microdot import MUTED_SOCKET_ERRORS
|
|
|
|
|
|
def _iscoroutine(coro):
|
|
return hasattr(coro, 'send') and hasattr(coro, 'throw')
|
|
|
|
|
|
class _AsyncBytesIO:
|
|
def __init__(self, data):
|
|
self.stream = io.BytesIO(data)
|
|
|
|
async def read(self, n=-1):
|
|
return self.stream.read(n)
|
|
|
|
async def readline(self): # pragma: no cover
|
|
return self.stream.readline()
|
|
|
|
async def readexactly(self, n): # pragma: no cover
|
|
return self.stream.read(n)
|
|
|
|
async def readuntil(self, separator=b'\n'): # pragma: no cover
|
|
return self.stream.readuntil(separator=separator)
|
|
|
|
async def awrite(self, data): # pragma: no cover
|
|
return self.stream.write(data)
|
|
|
|
async def aclose(self): # pragma: no cover
|
|
pass
|
|
|
|
|
|
class Request(BaseRequest):
|
|
@staticmethod
|
|
async def create(app, client_reader, client_writer, client_addr):
|
|
"""Create a request object.
|
|
|
|
:param app: The Microdot application instance.
|
|
:param client_reader: An input stream from where the request data can
|
|
be read.
|
|
:param client_writer: An output stream where the response data can be
|
|
written.
|
|
:param client_addr: The address of the client, as a tuple.
|
|
|
|
This method is a coroutine. It returns a newly created ``Request``
|
|
object.
|
|
"""
|
|
# request line
|
|
line = (await Request._safe_readline(client_reader)).strip().decode()
|
|
if not line:
|
|
return None
|
|
method, url, http_version = line.split()
|
|
http_version = http_version.split('/', 1)[1]
|
|
|
|
# headers
|
|
headers = NoCaseDict()
|
|
content_length = 0
|
|
while True:
|
|
line = (await Request._safe_readline(
|
|
client_reader)).strip().decode()
|
|
if line == '':
|
|
break
|
|
header, value = line.split(':', 1)
|
|
value = value.strip()
|
|
headers[header] = value
|
|
if header.lower() == 'content-length':
|
|
content_length = int(value)
|
|
|
|
# body
|
|
body = b''
|
|
if content_length and content_length <= Request.max_body_length:
|
|
body = await client_reader.readexactly(content_length)
|
|
stream = None
|
|
else:
|
|
body = b''
|
|
stream = client_reader
|
|
|
|
return Request(app, client_addr, method, url, http_version, headers,
|
|
body=body, stream=stream,
|
|
sock=(client_reader, client_writer))
|
|
|
|
@property
|
|
def stream(self):
|
|
if self._stream is None:
|
|
self._stream = _AsyncBytesIO(self._body)
|
|
return self._stream
|
|
|
|
@staticmethod
|
|
async def _safe_readline(stream):
|
|
line = (await stream.readline())
|
|
if len(line) > Request.max_readline:
|
|
raise ValueError('line too long')
|
|
return line
|
|
|
|
|
|
class Response(BaseResponse):
|
|
"""An HTTP response class.
|
|
|
|
:param body: The body of the response. If a dictionary or list is given,
|
|
a JSON formatter is used to generate the body. If a file-like
|
|
object or an async generator is given, a streaming response is
|
|
used. If a string is given, it is encoded from UTF-8. Else,
|
|
the body should be a byte sequence.
|
|
:param status_code: The numeric HTTP status code of the response. The
|
|
default is 200.
|
|
:param headers: A dictionary of headers to include in the response.
|
|
:param reason: A custom reason phrase to add after the status code. The
|
|
default is "OK" for responses with a 200 status code and
|
|
"N/A" for any other status codes.
|
|
"""
|
|
|
|
async def write(self, stream):
|
|
self.complete()
|
|
|
|
try:
|
|
# status code
|
|
reason = self.reason if self.reason is not None else \
|
|
('OK' if self.status_code == 200 else 'N/A')
|
|
await stream.awrite('HTTP/1.0 {status_code} {reason}\r\n'.format(
|
|
status_code=self.status_code, reason=reason).encode())
|
|
|
|
# headers
|
|
for header, value in self.headers.items():
|
|
values = value if isinstance(value, list) else [value]
|
|
for value in values:
|
|
await stream.awrite('{header}: {value}\r\n'.format(
|
|
header=header, value=value).encode())
|
|
await stream.awrite(b'\r\n')
|
|
|
|
# body
|
|
async for body in self.body_iter():
|
|
if isinstance(body, str): # pragma: no cover
|
|
body = body.encode()
|
|
await stream.awrite(body)
|
|
except OSError as exc: # pragma: no cover
|
|
if exc.errno in MUTED_SOCKET_ERRORS or \
|
|
exc.args[0] == 'Connection lost':
|
|
pass
|
|
else:
|
|
raise
|
|
|
|
def body_iter(self):
|
|
if hasattr(self.body, '__anext__'):
|
|
# response body is an async generator
|
|
return self.body
|
|
|
|
response = self
|
|
|
|
class iter:
|
|
def __aiter__(self):
|
|
if response.body:
|
|
self.i = 0 # need to determine type of response.body
|
|
else:
|
|
self.i = -1 # no response body
|
|
return self
|
|
|
|
async def __anext__(self):
|
|
if self.i == -1:
|
|
raise StopAsyncIteration
|
|
if self.i == 0:
|
|
if hasattr(response.body, 'read'):
|
|
self.i = 2 # response body is a file-like object
|
|
elif hasattr(response.body, '__next__'):
|
|
self.i = 1 # response body is a sync generator
|
|
return next(response.body)
|
|
else:
|
|
self.i = -1 # response body is a plain string
|
|
return response.body
|
|
elif self.i == 1:
|
|
try:
|
|
return next(response.body)
|
|
except StopIteration:
|
|
raise StopAsyncIteration
|
|
buf = response.body.read(response.send_file_buffer_size)
|
|
if _iscoroutine(buf): # pragma: no cover
|
|
buf = await buf
|
|
if len(buf) < response.send_file_buffer_size:
|
|
self.i = -1
|
|
if hasattr(response.body, 'close'): # pragma: no cover
|
|
result = response.body.close()
|
|
if _iscoroutine(result):
|
|
await result
|
|
return buf
|
|
|
|
return iter()
|
|
|
|
|
|
class Microdot(BaseMicrodot):
|
|
async def start_server(self, host='0.0.0.0', port=5000, debug=False,
|
|
ssl=None):
|
|
"""Start the Microdot web server as a coroutine. This coroutine does
|
|
not normally return, as the server enters an endless listening loop.
|
|
The :func:`shutdown` function provides a method for terminating the
|
|
server gracefully.
|
|
|
|
:param host: The hostname or IP address of the network interface that
|
|
will be listening for requests. A value of ``'0.0.0.0'``
|
|
(the default) indicates that the server should listen for
|
|
requests on all the available interfaces, and a value of
|
|
``127.0.0.1`` indicates that the server should listen
|
|
for requests only on the internal networking interface of
|
|
the host.
|
|
:param port: The port number to listen for requests. The default is
|
|
port 5000.
|
|
:param debug: If ``True``, the server logs debugging information. The
|
|
default is ``False``.
|
|
:param ssl: An ``SSLContext`` instance or ``None`` if the server should
|
|
not use TLS. The default is ``None``.
|
|
|
|
This method is a coroutine.
|
|
|
|
Example::
|
|
|
|
import asyncio
|
|
from microdot_asyncio import Microdot
|
|
|
|
app = Microdot()
|
|
|
|
@app.route('/')
|
|
async def index():
|
|
return 'Hello, world!'
|
|
|
|
async def main():
|
|
await app.start_server(debug=True)
|
|
|
|
asyncio.run(main())
|
|
"""
|
|
self.debug = debug
|
|
|
|
async def serve(reader, writer):
|
|
if not hasattr(writer, 'awrite'): # pragma: no cover
|
|
# CPython provides the awrite and aclose methods in 3.8+
|
|
async def awrite(self, data):
|
|
self.write(data)
|
|
await self.drain()
|
|
|
|
async def aclose(self):
|
|
self.close()
|
|
await self.wait_closed()
|
|
|
|
from types import MethodType
|
|
writer.awrite = MethodType(awrite, writer)
|
|
writer.aclose = MethodType(aclose, writer)
|
|
|
|
await self.handle_request(reader, writer)
|
|
|
|
if self.debug: # pragma: no cover
|
|
print('Starting async server on {host}:{port}...'.format(
|
|
host=host, port=port))
|
|
|
|
try:
|
|
self.server = await asyncio.start_server(serve, host, port,
|
|
ssl=ssl)
|
|
except TypeError:
|
|
self.server = await asyncio.start_server(serve, host, port)
|
|
|
|
while True:
|
|
try:
|
|
await self.server.wait_closed()
|
|
break
|
|
except AttributeError: # pragma: no cover
|
|
# the task hasn't been initialized in the server object yet
|
|
# wait a bit and try again
|
|
await asyncio.sleep(0.1)
|
|
|
|
def run(self, host='0.0.0.0', port=5000, debug=False, ssl=None):
|
|
"""Start the web server. This function does not normally return, as
|
|
the server enters an endless listening loop. The :func:`shutdown`
|
|
function provides a method for terminating the server gracefully.
|
|
|
|
:param host: The hostname or IP address of the network interface that
|
|
will be listening for requests. A value of ``'0.0.0.0'``
|
|
(the default) indicates that the server should listen for
|
|
requests on all the available interfaces, and a value of
|
|
``127.0.0.1`` indicates that the server should listen
|
|
for requests only on the internal networking interface of
|
|
the host.
|
|
:param port: The port number to listen for requests. The default is
|
|
port 5000.
|
|
:param debug: If ``True``, the server logs debugging information. The
|
|
default is ``False``.
|
|
:param ssl: An ``SSLContext`` instance or ``None`` if the server should
|
|
not use TLS. The default is ``None``.
|
|
|
|
Example::
|
|
|
|
from microdot_asyncio import Microdot
|
|
|
|
app = Microdot()
|
|
|
|
@app.route('/')
|
|
async def index():
|
|
return 'Hello, world!'
|
|
|
|
app.run(debug=True)
|
|
"""
|
|
asyncio.run(self.start_server(host=host, port=port, debug=debug,
|
|
ssl=ssl))
|
|
|
|
def shutdown(self):
|
|
self.server.close()
|
|
|
|
async def handle_request(self, reader, writer):
|
|
req = None
|
|
try:
|
|
req = await Request.create(self, reader, writer,
|
|
writer.get_extra_info('peername'))
|
|
except Exception as exc: # pragma: no cover
|
|
print_exception(exc)
|
|
|
|
res = await self.dispatch_request(req)
|
|
if res != Response.already_handled: # pragma: no branch
|
|
await res.write(writer)
|
|
try:
|
|
await writer.aclose()
|
|
except OSError as exc: # pragma: no cover
|
|
if exc.errno in MUTED_SOCKET_ERRORS:
|
|
pass
|
|
else:
|
|
raise
|
|
if self.debug and req: # pragma: no cover
|
|
print('{method} {path} {status_code}'.format(
|
|
method=req.method, path=req.path,
|
|
status_code=res.status_code))
|
|
|
|
async def dispatch_request(self, req):
|
|
if req:
|
|
if req.content_length > req.max_content_length:
|
|
if 413 in self.error_handlers:
|
|
res = await self._invoke_handler(
|
|
self.error_handlers[413], req)
|
|
else:
|
|
res = 'Payload too large', 413
|
|
else:
|
|
f = self.find_route(req)
|
|
try:
|
|
res = None
|
|
if callable(f):
|
|
for handler in self.before_request_handlers:
|
|
res = await self._invoke_handler(handler, req)
|
|
if res:
|
|
break
|
|
if res is None:
|
|
res = await self._invoke_handler(
|
|
f, req, **req.url_args)
|
|
if isinstance(res, tuple):
|
|
body = res[0]
|
|
if isinstance(res[1], int):
|
|
status_code = res[1]
|
|
headers = res[2] if len(res) > 2 else {}
|
|
else:
|
|
status_code = 200
|
|
headers = res[1]
|
|
res = Response(body, status_code, headers)
|
|
elif not isinstance(res, Response):
|
|
res = Response(res)
|
|
for handler in self.after_request_handlers:
|
|
res = await self._invoke_handler(
|
|
handler, req, res) or res
|
|
for handler in req.after_request_handlers:
|
|
res = await handler(req, res) or res
|
|
elif f in self.error_handlers:
|
|
res = await self._invoke_handler(
|
|
self.error_handlers[f], req)
|
|
else:
|
|
res = 'Not found', f
|
|
except HTTPException as exc:
|
|
if exc.status_code in self.error_handlers:
|
|
res = self.error_handlers[exc.status_code](req)
|
|
else:
|
|
res = exc.reason, exc.status_code
|
|
except Exception as exc:
|
|
print_exception(exc)
|
|
exc_class = None
|
|
res = None
|
|
if exc.__class__ in self.error_handlers:
|
|
exc_class = exc.__class__
|
|
else:
|
|
for c in mro(exc.__class__)[1:]:
|
|
if c in self.error_handlers:
|
|
exc_class = c
|
|
break
|
|
if exc_class:
|
|
try:
|
|
res = await self._invoke_handler(
|
|
self.error_handlers[exc_class], req, exc)
|
|
except Exception as exc2: # pragma: no cover
|
|
print_exception(exc2)
|
|
if res is None:
|
|
if 500 in self.error_handlers:
|
|
res = await self._invoke_handler(
|
|
self.error_handlers[500], req)
|
|
else:
|
|
res = 'Internal server error', 500
|
|
else:
|
|
if 400 in self.error_handlers:
|
|
res = await self._invoke_handler(self.error_handlers[400], req)
|
|
else:
|
|
res = 'Bad request', 400
|
|
if isinstance(res, tuple):
|
|
res = Response(*res)
|
|
elif not isinstance(res, Response):
|
|
res = Response(res)
|
|
return res
|
|
|
|
async def _invoke_handler(self, f_or_coro, *args, **kwargs):
|
|
ret = f_or_coro(*args, **kwargs)
|
|
if _iscoroutine(ret):
|
|
ret = await ret
|
|
return ret
|
|
|
|
|
|
abort = Microdot.abort
|
|
Response.already_handled = Response()
|
|
redirect = Response.redirect
|
|
send_file = Response.send_file
|