Function _webio_handler

pywebio/platform/aiohttp.py:74–125

: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 :return: aiohttp Request Handler

(applications, cdn, websocket_settings, reconnect_timeout=0, check_origin_func=_is_same_site)

def _webio_handler(applications, cdn, websocket_settings, reconnect_timeout=0, check_origin_func=_is_same_site):
    """
    :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
    :return: aiohttp Request Handler
    """
    ws_adaptor.set_expire_second(reconnect_timeout)

    async def wshandle(request: web.Request):
        ioloop = asyncio.get_event_loop()
        asyncio.get_event_loop().create_task(ws_adaptor.session_clean_task())

        origin = request.headers.get('origin')
        if origin and not check_origin_func(origin=origin, host=request.host):
            return web.Response(status=403, text="Cross origin websockets not allowed")

        if request.headers.get("Upgrade", "").lower() != "websocket":
            # Backward compatible
            if request.query.getone('test', ''):
                return web.Response(text="")

            app_name = request.query.getone('app', 'index')
            app = applications.get(app_name) or applications['index']
            no_cdn = cdn is True and request.query.getone('_pywebio_cdn', '') == 'false'
            html = render_page(app, protocol='ws', cdn=False if no_cdn else cdn)
            return web.Response(body=html, content_type='text/html')

        ws = web.WebSocketResponse(**websocket_settings)
        await ws.prepare(request)

        app_name = request.query.getone('app', 'index')
        application = applications.get(app_name) or applications['index']

        conn = WebSocketConnection(ws, request, ioloop)
        handler = ws_adaptor.WebSocketHandler(
            connection=conn, application=application, reconnectable=bool(reconnect_timeout), ioloop=ioloop
        )

        # see: https://github.com/aio-libs/aiohttp/issues/1768
        try:
            async for msg in ws:
                if msg.type in (web.WSMsgType.text, web.WSMsgType.binary):
                    handler.send_client_data(msg.data)
                elif msg.type == web.WSMsgType.close:
                    raise asyncio.CancelledError()
        finally:
            handler.notify_connection_lost()

        return ws

    return wshandle


def webio_handler(applications, cdn=True, reconnect_timeout=0, allowed_origins=None, check_origin=None,