import os
import json
import logging
import tornado.ioloop
import tornado.web
from . import page
from ..session import Session
from .adaptor.http import HttpContext, HttpHandler
from .tornado import set_ioloop, _setup_server, open_webbrowser_on_server_started
from .utils import cdn_validation, print_listen_address
from ..utils import parse_file_size
logger = logging.getLogger(__name__)
class TornadoHttpContext(HttpContext):
backend_name = 'tornado'
def __init__(self, handler: tornado.web.RequestHandler):
self.handler = handler
self.response = b''
def request_obj(self):
"""返回当前请求对象"""
return self.handler.request
def request_method(self):
"""返回当前请求的方法，大写"""
return self.handler.request.method.upper()
def request_headers(self):
"""返回当前请求的header字典"""
return self.handler.request.headers
def request_url_parameter(self, name, default=None):
"""返回当前请求的URL参数"""
return self.handler.get_query_argument(name, default=default)
def request_body(self):
return self.handler.request.body
def set_header(self, name, value):
"""为当前响应设置header"""
self.handler.set_header(name, value)
def set_status(self, status: int):
"""为当前响应设置http status"""
self.handler.set_status(status)
def set_content(self, content, json_type=False):
"""设置相应的内容
:param content:
:param bool json_type: content是否要序列化成json格式，并将 content-type 设置为application/json
"""
# self.response.content accept str and byte
if json_type:
self.set_header('content-type', 'application/json')
self.response = json.dumps(content)
else:
self.response = content
def get_response(self):
"""获取当前的响应对象，用于在私图函数中返回"""
return self.response
def get_client_ip(self):
"""获取用户的ip"""
return self.handler.request.remote_ip
def get_path(self):
"""Get the path patton of the http request uri"""
return self.handler.request.path
[docs]def webio_handler(applications, cdn=True,
session_expire_seconds=None,
session_cleanup_interval=None,
allowed_origins=None, check_origin=None):
"""Get the ``RequestHandler`` class for running PyWebIO applications in Tornado.
The ``RequestHandler`` communicates with the browser by HTTP protocol.
The arguments of ``webio_handler()`` have the same meaning as for :func:`pywebio.platform.tornado_http.start_server`
.. versionadded:: 1.2
"""
cdn = cdn_validation(cdn, 'error') # if CDN is not available, raise error
handler = HttpHandler(applications=applications, cdn=cdn,
session_expire_seconds=session_expire_seconds,
session_cleanup_interval=session_cleanup_interval,
allowed_origins=allowed_origins, check_origin=check_origin)
class MainHandler(tornado.web.RequestHandler):
def options(self):
return self.get()
def post(self):
return self.get()
async def get(self):
context = TornadoHttpContext(self)
response = await handler.handle_request_async(context)
self.write(response)
return MainHandler
[docs]def start_server(applications, port=8080, host='',
debug=False, cdn=True, static_dir=None,
allowed_origins=None, check_origin=None,
auto_open_webbrowser=False,
session_expire_seconds=None,
session_cleanup_interval=None,
max_payload_size='200M',
**tornado_app_settings):
"""Start a Tornado server to provide the PyWebIO application as a web service.
The Tornado server communicates with the browser by HTTP protocol.
:param int session_expire_seconds: Session expiration time, in seconds(default 60s).
If no client message is received within ``session_expire_seconds``, the session will be considered expired.
:param int session_cleanup_interval: Session cleanup interval, in seconds(default 120s).
The server will periodically clean up expired sessions and release the resources occupied by the sessions.
:param int/str max_payload_size: Max size of a request body which Tornado can accept.
The rest arguments of ``start_server()`` have the same meaning as for :func:`pywebio.platform.tornado.start_server`
.. versionadded:: 1.2
"""
if not host:
host = '0.0.0.0'
cdn = cdn_validation(cdn, 'warn')
set_ioloop(tornado.ioloop.IOLoop.current()) # to enable bokeh app
cdn = cdn_validation(cdn, 'warn') # if CDN is not available, warn user and disable CDN
page.MAX_PAYLOAD_SIZE = max_payload_size = parse_file_size(max_payload_size)
debug = Session.debug = os.environ.get('PYWEBIO_DEBUG', debug)
tornado_app_settings.setdefault('websocket_max_message_size', max_payload_size)
tornado_app_settings['websocket_max_message_size'] = parse_file_size(
tornado_app_settings['websocket_max_message_size'])
tornado_app_settings['debug'] = debug
handler = webio_handler(applications, cdn,
session_expire_seconds=session_expire_seconds,
session_cleanup_interval=session_cleanup_interval,
allowed_origins=allowed_origins, check_origin=check_origin)
_, port = _setup_server(webio_handler=handler, port=port, host=host, static_dir=static_dir,
max_buffer_size=parse_file_size(max_payload_size), **tornado_app_settings)
print_listen_address(host, port)
if auto_open_webbrowser:
tornado.ioloop.IOLoop.current().spawn_callback(open_webbrowser_on_server_started, host or '127.0.0.1', port)
tornado.ioloop.IOLoop.current().start()