import asyncio
import json
import logging
from functools import partial
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute, Mount
from starlette.websockets import WebSocket
from starlette.websockets import WebSocketDisconnect
from .remote_access import start_remote_access_service
from .tornado import open_webbrowser_on_server_started
from .utils import make_applications, render_page, cdn_validation, OriginChecker, deserialize_binary_event
from ..session import CoroutineBasedSession, ThreadBasedSession, register_session_implement_for_target, Session
from ..session.base import get_session_info_from_headers
from ..utils import get_free_port, STATIC_PATH, iscoroutinefunction, isgeneratorfunction, strip_space
logger = logging.getLogger(__name__)
def _webio_routes(applications, cdn, check_origin_func):
"""
:param dict applications: dict of `name -> task function`
:param bool/str cdn: Whether to load front-end static resources from CDN
:param callable check_origin_func: check_origin_func(origin, host) -> bool
"""
async def http_endpoint(request: Request):
origin = request.headers.get('origin')
if origin and not check_origin_func(origin=origin, host=request.headers.get('host')):
return HTMLResponse(status_code=403, content="Cross origin websockets not allowed")
# Backward compatible
if request.query_params.get('test'):
return HTMLResponse(content="")
app_name = request.query_params.get('app', 'index')
app = applications.get(app_name) or applications['index']
html = render_page(app, protocol='ws', cdn=cdn)
return HTMLResponse(content=html)
async def websocket_endpoint(websocket: WebSocket):
ioloop = asyncio.get_event_loop()
await websocket.accept()
close_from_session_tag = False # session close causes websocket close
def send_msg_to_client(session: Session):
for msg in session.get_task_commands():
ioloop.create_task(websocket.send_json(msg))
def close_from_session():
nonlocal close_from_session_tag
close_from_session_tag = True
ioloop.create_task(websocket.close())
logger.debug("WebSocket closed from session")
session_info = get_session_info_from_headers(websocket.headers)
session_info['user_ip'] = websocket.client.host or ''
session_info['request'] = websocket
session_info['backend'] = 'starlette'
session_info['protocol'] = 'websocket'
app_name = websocket.query_params.get('app', 'index')
application = applications.get(app_name) or applications['index']
if iscoroutinefunction(application) or isgeneratorfunction(application):
session = CoroutineBasedSession(application, session_info=session_info,
on_task_command=send_msg_to_client,
on_session_close=close_from_session)
else:
session = ThreadBasedSession(application, session_info=session_info,
on_task_command=send_msg_to_client,
on_session_close=close_from_session, loop=ioloop)
while True:
try:
msg = await websocket.receive()
if msg["type"] == "websocket.disconnect":
raise WebSocketDisconnect(msg["code"])
text, binary = msg.get('text'), msg.get('bytes')
event = None
if text:
event = json.loads(text)
elif binary:
event = deserialize_binary_event(binary)
except WebSocketDisconnect:
if not close_from_session_tag:
# close session because client disconnected to server
session.close(nonblock=True)
logger.debug("WebSocket closed from client")
break
if event is not None:
session.send_client_event(event)
return [
Route("/", http_endpoint),
WebSocketRoute("/", websocket_endpoint)
]
[docs]def webio_routes(applications, cdn=True, allowed_origins=None, check_origin=None):
"""Get the FastAPI/Starlette routes for running PyWebIO applications.
The API communicates with the browser using WebSocket protocol.
The arguments of ``webio_routes()`` have the same meaning as for :func:`pywebio.platform.fastapi.start_server`
.. versionadded:: 1.3
:return: FastAPI/Starlette routes
"""
try:
import websockets
except Exception:
raise RuntimeError(strip_space("""
Missing dependency package `websockets` for websocket support.
You can install it with the following command:
pip install websockets
""".strip(), n=8))
applications = make_applications(applications)
for target in applications.values():
register_session_implement_for_target(target)
cdn = cdn_validation(cdn, 'error')
if check_origin is None:
check_origin_func = partial(OriginChecker.check_origin, allowed_origins=allowed_origins or [])
else:
check_origin_func = lambda origin, host: OriginChecker.is_same_site(origin, host) or check_origin(origin)
return _webio_routes(applications=applications, cdn=cdn, check_origin_func=check_origin_func)
[docs]def start_server(applications, port=0, host='', cdn=True,
static_dir=None, remote_access=False, debug=False,
allowed_origins=None, check_origin=None,
auto_open_webbrowser=False,
**uvicorn_settings):
"""Start a FastAPI/Starlette server using uvicorn to provide the PyWebIO application as a web service.
:param bool debug: Boolean indicating if debug tracebacks should be returned on errors.
:param uvicorn_settings: Additional keyword arguments passed to ``uvicorn.run()``.
For details, please refer: https://www.uvicorn.org/settings/
The rest arguments of ``start_server()`` have the same meaning as for :func:`pywebio.platform.tornado.start_server`
.. versionadded:: 1.3
"""
app = asgi_app(applications, cdn=cdn, static_dir=static_dir, debug=debug,
allowed_origins=allowed_origins, check_origin=check_origin)
if auto_open_webbrowser:
asyncio.get_event_loop().create_task(open_webbrowser_on_server_started('localhost', port))
if not host:
host = '0.0.0.0'
if port == 0:
port = get_free_port()
if remote_access or remote_access == {}:
if remote_access is True: remote_access = {}
start_remote_access_service(**remote_access, local_port=port)
uvicorn.run(app, host=host, port=port, **uvicorn_settings)
[docs]def asgi_app(applications, cdn=True, static_dir=None, debug=False, allowed_origins=None, check_origin=None):
"""Get the starlette/Fastapi ASGI app for running PyWebIO applications.
Use :func:`pywebio.platform.fastapi.webio_routes` if you prefer handling static files yourself.
The arguments of ``asgi_app()`` have the same meaning as for :func:`pywebio.platform.fastapi.start_server`
:Example:
To be used with ``FastAPI.mount()`` to include pywebio as a subapp into an existing Starlette/FastAPI application::
from fastapi import FastAPI
from pywebio.platform.fastapi import asgi_app
from pywebio.output import put_text
app = FastAPI()
subapp = asgi_app(lambda: put_text("hello from pywebio"))
app.mount("/pywebio", subapp)
:Returns: Starlette/Fastapi ASGI app
.. versionadded:: 1.3
"""
try:
from starlette.staticfiles import StaticFiles
except Exception:
raise RuntimeError(strip_space("""
Missing dependency package `aiofiles` for static file serving.
You can install it with the following command:
pip install aiofiles
""".strip(), n=8))
cdn = cdn_validation(cdn, 'warn')
if cdn is False:
cdn = 'pywebio_static'
routes = webio_routes(applications, cdn=cdn, allowed_origins=allowed_origins, check_origin=check_origin)
if static_dir:
routes.append(Mount('/static', app=StaticFiles(directory=static_dir), name="static"))
routes.append(Mount('/pywebio_static', app=StaticFiles(directory=STATIC_PATH), name="pywebio_static"))
return Starlette(routes=routes, debug=debug)