This section will introduce the advanced features of PyWebIO.
start_server() <pywebio.platform.tornado.start_server> accepts a function as PyWebIO application. In addition, start_server() <pywebio.platform.tornado.start_server> also accepts a list of application function or a dictionary of it to start multiple applications. You can use pywebio.session.go_app() <pywebio.session.go_app> or put_link() <pywebio.output.put_link> to jump between application:
def task_1():
put_text('task_1')
put_buttons(['Go task 2'], [lambda: go_app('task_2')])
def task_2():
put_text('task_2')
put_buttons(['Go task 1'], [lambda: go_app('task_1')])
def index():
put_link('Go task 1', app='task_1') # Use `app` parameter to specify the task name
put_link('Go task 2', app='task_2')
# equal to `start_server({'index': index, 'task_1': task_1, 'task_2': task_2})`
start_server([index, task_1, task_2])
When the first parameter of start_server() <pywebio.platform.tornado.start_server> is a dictionary, whose key is application name and value is application function. When it is a list, PyWebIO will use function name as application name.
You can select which application to access through the app URL parameter
(for example, visit http://host:port/?app=foo to access the foo application),
By default, the index application is opened when no app URL parameter provided.
When the index application doesn't exist, PyWebIO will provide a default index application.
The PyWebIO application can be integrated into an existing Python Web project, the PyWebIO application and the Web project share a web framework. PyWebIO currently supports integration with Flask, Tornado, Django, aiohttp and FastAPI(Starlette) web frameworks.
The integration methods of those web frameworks are as follows:
.. tabs::
.. tab:: Tornado
.. only:: latex
**Tornado**
Use `pywebio.platform.tornado.webio_handler()` to get the
`WebSocketHandler <https://www.tornadoweb.org/en/stable/websocket.html#tornado.websocket.WebSocketHandler>`_
class for running PyWebIO applications in Tornado::
import tornado.ioloop
import tornado.web
from pywebio.platform.tornado import webio_handler
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
if __name__ == "__main__":
application = tornado.web.Application([
(r"/", MainHandler),
(r"/tool", webio_handler(task_func)), # `task_func` is PyWebIO task function
])
application.listen(port=80, address='localhost')
tornado.ioloop.IOLoop.current().start()
In above code, we add a routing rule to bind the ``WebSocketHandler`` of the PyWebIO application to the ``/tool`` path.
After starting the Tornado server, you can visit ``http://localhost/tool`` to open the PyWebIO application.
.. attention::
PyWebIO uses the WebSocket protocol to communicate with the browser in Tornado. If your Tornado application
is behind a reverse proxy (such as Nginx), you may need to configure the reverse proxy to support the
WebSocket protocol. :ref:`Here <nginx_ws_config>` is an example of Nginx WebSocket configuration.
.. tab:: Flask
.. only:: latex
**Flask**
Use `pywebio.platform.flask.webio_view()` to get the view function for running PyWebIO applications in Flask::
from pywebio.platform.flask import webio_view
from flask import Flask
app = Flask(__name__)
# `task_func` is PyWebIO task function
app.add_url_rule('/tool', 'webio_view', webio_view(task_func),
methods=['GET', 'POST', 'OPTIONS']) # need GET,POST and OPTIONS methods
app.run(host='localhost', port=80)
In above code, we add a routing rule to bind the view function of the PyWebIO application to the ``/tool`` path.
After starting the Flask application, visit ``http://localhost/tool`` to open the PyWebIO application.
.. tab:: Django
.. only:: latex
**Django**
Use `pywebio.platform.django.webio_view()` to get the view function for running PyWebIO applications in Django::
# urls.py
from django.urls import path
from pywebio.platform.django import webio_view
# `task_func` is PyWebIO task function
webio_view_func = webio_view(task_func)
urlpatterns = [
path(r"tool", webio_view_func),
]
In above code, we add a routing rule to bind the view function of the PyWebIO application to the ``/tool`` path.
After starting the Django server, visit ``http://localhost/tool`` to open the PyWebIO application
.. tab:: aiohttp
.. only:: latex
**aiohttp**
Use `pywebio.platform.aiohttp.webio_handler()` to get the
`Request Handler <https://docs.aiohttp.org/en/stable/web_quickstart.html#aiohttp-web-handler>`_ coroutine for
running PyWebIO applications in aiohttp::
from aiohttp import web
from pywebio.platform.aiohttp import webio_handler
app = web.Application()
# `task_func` is PyWebIO task function
app.add_routes([web.get('/tool', webio_handler(task_func))])
web.run_app(app, host='localhost', port=80)
After starting the aiohttp server, visit ``http://localhost/tool`` to open the PyWebIO application
.. attention::
PyWebIO uses the WebSocket protocol to communicate with the browser in aiohttp. If your aiohttp server is
behind a reverse proxy (such as Nginx), you may need to configure the reverse proxy to support the WebSocket
protocol. :ref:`Here <nginx_ws_config>` is an example of Nginx WebSocket configuration.
.. tab:: FastAPI/Starlette
.. only:: latex
**FastAPI/Starlette**
Use `pywebio.platform.fastapi.webio_routes()` to get the FastAPI/Starlette routes for running PyWebIO applications.
You can mount the routes to your FastAPI/Starlette app.
FastAPI::
from fastapi import FastAPI
from pywebio.platform.fastapi import webio_routes
app = FastAPI()
@app.get("/app")
def read_main():
return {"message": "Hello World from main app"}
# `task_func` is PyWebIO task function
app.mount("/tool", FastAPI(routes=webio_routes(task_func)))
Starlette::
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route, Mount
from pywebio.platform.fastapi import webio_routes
async def homepage(request):
return JSONResponse({'hello': 'world'})
app = Starlette(routes=[
Route('/', homepage),
Mount('/tool', routes=webio_routes(task_func)) # `task_func` is PyWebIO task function
])
After starting the server by using ``uvicorn <module>:app`` , visit ``http://localhost:8000/tool/`` to open the PyWebIO application
See also: `FastAPI doc <https://fastapi.tiangolo.com/advanced/sub-applications/>`_ , `Starlette doc <https://www.starlette.io/routing/#submounting-routes>`_
.. attention::
PyWebIO uses the WebSocket protocol to communicate with the browser in FastAPI/Starlette. If your server is
behind a reverse proxy (such as Nginx), you may need to configure the reverse proxy to support the WebSocket
protocol. :ref:`Here <nginx_ws_config>` is an example of Nginx WebSocket configuration.
Deployment in production
In your production system, you may want to deploy the web applications with some WSGI/ASGI servers such as uWSGI, Gunicorn, and Uvicorn. Since PyWebIO applications store session state in memory of process, when you use HTTP-based sessions (Flask and Django) and spawn multiple workers to handle requests, the request may be dispatched to a process that does not hold the session to which the request belongs. So you can only start one worker to handle requests when using Flask or Django backend.
If you still want to use multiple processes to increase concurrency, one way is to use Uvicorn+FastAPI, or you can also start multiple Tornado/aiohttp processes and add external load balancer (such as HAProxy or nginx) before them. Those backends use the WebSocket protocol to communicate with the browser in PyWebIO, so there is no the issue as described above.
Static resources Hosting
By default, the front-end of PyWebIO gets required static resources from CDN. If you want to deploy PyWebIO applications
in an offline environment, you need to host static files by yourself, and set the cdn parameter of webio_view()
or webio_handler() to False.
When setting cdn=False , you need to host the static resources in the same directory as the PyWebIO application.
In addition, you can also pass a string to cdn parameter to directly set the URL of PyWebIO static resources directory.
The path of the static file of PyWebIO is stored in pywebio.STATIC_PATH, you can use the command
python3 -c "import pywebio; print(pywebio.STATIC_PATH)" to print it out.
Note
start_server() and path_deploy() also support cdn parameter, if it is set to False, the static
resource will be hosted in local server automatically, without manual hosting.
In most cases, you don’t need the coroutine-based session. All functions or methods in PyWebIO that are only used for coroutine sessions are specifically noted in the document.
PyWebIO's session is based on thread by default. Each time a user opens a session connection to the server, PyWebIO will start a thread to run the task function. In addition to thread-based sessions, PyWebIO also provides coroutine-based sessions. Coroutine-based sessions accept coroutine functions as task functions.
The session based on the coroutine is a single-thread model, which means that all sessions run in a single thread. For IO-bound tasks, coroutines take up fewer resources than threads and have performance comparable to threads. In addition, the context switching of the coroutine is predictable, which can reduce the need for program synchronization and locking, and can effectively avoid most critical section problems.
To use coroutine-based session, you need to use the async keyword to declare the task function as a coroutine
function, and use the await syntax to call the PyWebIO input function:
from pywebio.input import *
from pywebio.output import *
from pywebio import start_server
async def say_hello():
name = await input("what's your name?")
put_text('Hello, %s' % name)
start_server(say_hello, auto_open_webbrowser=True)In the coroutine task function, you can also use await to call other coroutines or
(awaitable objects) in the standard
library asyncio:
import asyncio
from pywebio import start_server
async def hello_word():
put_text('Hello ...')
await asyncio.sleep(1) # await awaitable objects in asyncio
put_text('... World!')
async def main():
await hello_word() # await coroutine
put_text('Bye, bye')
start_server(main, auto_open_webbrowser=True)Attention!
In coroutine-based session, all input functions defined in the :doc:`pywebio.input </input>` module need to use
await syntax to get the return value. Forgetting to use await will be a common error when using coroutine-based session.
Other functions that need to use await syntax in the coroutine session are:
- pywebio.session.run_asyncio_coroutine(coro_obj) <pywebio.session.run_asyncio_coroutine>
- pywebio.session.eval_js(expression) <pywebio.session.eval_js>
Warning
Although the PyWebIO coroutine session is compatible with the awaitable objects in the standard library asyncio,
the asyncio library is not compatible with the awaitable objects in the PyWebIO coroutine session.
That is to say, you can't pass PyWebIO awaitable objects to the asyncio functions that accept awaitable objects.
For example, the following calls are not supported
await asyncio.shield(pywebio.input())
await asyncio.gather(asyncio.sleep(1), pywebio.session.eval_js('1+1'))
task = asyncio.create_task(pywebio.input())
In coroutine-based session, you can start new thread, but you cannot call PyWebIO interactive functions in it (register_thread() <pywebio.session.register_thread> is not available in coroutine session). But you can use run_async(coro) <pywebio.session.run_async> to execute a coroutine object asynchronously, and PyWebIO interactive functions can be used in the new coroutine:
from pywebio import start_server
from pywebio.session import run_async
async def counter(n):
for i in range(n):
put_text(i)
await asyncio.sleep(1)
async def main():
run_async(counter(10))
put_text('Main coroutine function exited.')
start_server(main, auto_open_webbrowser=True)run_async(coro) <pywebio.session.run_async> returns a TaskHandler <pywebio.session.coroutinebased.TaskHandler>, which can be used to query the running status of the coroutine or close the coroutine.
Similar to thread-based session, when user close the browser page, the session will be closed.
After the browser page closed, PyWebIO input function calls that have not yet returned in the current session will cause SessionClosedException <pywebio.exceptions.SessionClosedException>, and subsequent calls to PyWebIO interactive functions will cause SessionNotFoundException <pywebio.exceptions.SessionNotFoundException> or SessionClosedException <pywebio.exceptions.SessionClosedException>.
defer_call(func) <pywebio.session.defer_call> also available in coroutine session.
The PyWebIO application that using coroutine-based session can also be integrated to the web framework.
However, there are some limitations when using coroutine-based sessions to integrate into Flask or Django:
First, when await the coroutine objects/awaitable objects in the asyncio module, you need to use
run_asyncio_coroutine() <pywebio.session.run_asyncio_coroutine> to wrap the coroutine object.
Secondly, you need to start a new thread to run the event loop before starting a Flask/Django server.
Example of coroutine-based session integration into Flask:
import asyncio
import threading
from flask import Flask, send_from_directory
from pywebio import STATIC_PATH
from pywebio.output import *
from pywebio.platform.flask import webio_view
from pywebio.platform import run_event_loop
from pywebio.session import run_asyncio_coroutine
async def hello_word():
put_text('Hello ...')
await run_asyncio_coroutine(asyncio.sleep(1)) # can't just "await asyncio.sleep(1)"
put_text('... World!')
app = Flask(__name__)
app.add_url_rule('/hello', 'webio_view', webio_view(hello_word),
methods=['GET', 'POST', 'OPTIONS'])
# thread to run event loop
threading.Thread(target=run_event_loop, daemon=True).start()
app.run(host='localhost', port=80)Finally, coroutine-based session is not available in the script mode. You always need to use start_server() to
run coroutine task function or integrate it to a web framework.