Advanced topic

This section will introduce the advanced features of PyWebIO.

Start multiple applications with start_server()

start_server() accepts a function as PyWebIO application. In addition, 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() or 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() 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.

Integration with web framework

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:

Use pywebio.platform.tornado.webio_handler() to get the 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. Here is an example of Nginx WebSocket configuration.

Notes

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.

Coroutine-based session

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.

Using coroutine session

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 pywebio.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:

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())

Concurrency in coroutine-based sessions

In coroutine-based session, you can start new thread, but you cannot call PyWebIO interactive functions in it (register_thread() is not available in coroutine session). But you can use run_async(coro) 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) returns a TaskHandler, which can be used to query the running status of the coroutine or close the coroutine.

Close of session

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, and subsequent calls to PyWebIO interactive functions will cause SessionNotFoundException or SessionClosedException.

defer_call(func) also available in coroutine session.

Integration with Web Framework

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() 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.