PyWebIO

https://pywebio.readthedocs.io/en/latest/
Admin
PyWebIO provides imperative functions to get user input and output content on the browser, turning...
Tokens:145,902
Snippets:721
Trust Score:8.3
Update:1 year ago
Show doc for...
Tokens:
Raw
### Setup Tornado Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Configures and sets up a Tornado web application. It defines request handlers, including static file serving, and starts the server on a specified port and host.

```Python
def _setup_server(webio_handler, port=0, host='', static_dir=None, max_buffer_size=2 ** 20 * 200,
                  **tornado_app_settings):
    if port == 0:
        port = get_free_port()

    handlers = [(r"/", webio_handler)]

    if static_dir is not None:
        handlers.append((r"/static/(.*)", tornado.web.StaticFileHandler, {"path": static_dir}))

    handlers.append((r"/(.*)", tornado.web.StaticFileHandler, {"path": STATIC_PATH, 'default_filename': 'index.html'}))

    app = tornado.web.Application(handlers=handlers, **tornado_app_settings)
    # Credit: https://stackoverflow.com/questions/19074972/content-length-too-long-when-uploading-file-using-tornado
    server = app.listen(port, address=host, max_buffer_size=max_buffer_size)
    return server, port
```

--------------------------------

### PyWebIO Server Mode - Multi-threading Example

Source: https://pywebio.readthedocs.io/en/latest/guide

Demonstrates using PyWebIO interactive functions within multiple threads in server mode. It shows how to register threads using `register_thread` to manage PyWebIO sessions correctly. An example of a background clock task is provided.

```Python
import threading
import time
import datetime
from pywebio import *
from pywebio.session import register_thread

def show_time():
    while True:
        with use_scope(name='time', clear=True):
            put_text(datetime.datetime.now())
            time.sleep(1)

def app():
    t = threading.Thread(target=show_time)
    register_thread(t)
    put_markdown('## Clock')
    t.start()  # run `show_time()` in background

    # ❌ this thread will cause `SessionNotFoundException`
    threading.Thread(target=show_time).start()

    put_text('Background task started.')


start_server(app, port=8080, debug=True)
```

--------------------------------

### Start PyWebIO Server

Source: https://pywebio.readthedocs.io/en/latest/platform

Starts a Python Web server to serve PyWebIO applications. Supports `cdn` and `static_dir` parameters for customization.

```Python
from pywebio import start_server

# Example usage of start_server
# start_server(your_pywebio_app, port=8080, cdn=True, static_dir='static/')

```

--------------------------------

### Configure and Start PyWebIO Tornado Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado_http

Sets up Tornado application settings, including WebSocket message size and debug mode. It then initializes the PyWebIO handler and starts the Tornado server. The server's listen address is printed, and a callback is spawned to open a web browser if auto_open_webbrowser is true. Finally, the Tornado IOLoop is started.

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

--------------------------------

### Install Aiohttp for PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/platform

Installs aiohttp version 3.1 or higher, which is necessary when using aiohttp as a PyWebIO backend server.

```bash
pip3 install -U aiohttp>=3.1
```

--------------------------------

### Install PyWebIO Battery

Source: https://pywebio.readthedocs.io/en/latest/battery

Instructions on how to install the PyWebIO battery extension package using pip.

```shell
pip3 install -U pywebio-battery
```

--------------------------------

### Install PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/index

Installs the stable version of PyWebIO using pip. Requires Python 3.5.2 or newer.

```bash
pip3 install -U pywebio
```

--------------------------------

### PyWebIO Output Handler Example

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/output

Demonstrates the usage of the `output` handler in PyWebIO for dynamically managing content. The example shows how to initialize, reset, append, and insert content into an output handler.

```python
@safely_destruct_output_when_exp('contents')
def output(*contents):
    """Placeholder of output

    .. deprecated:: 1.5
        See :ref:`User Guide <put_scope>` for new way to set css style for output.

     ``output()`` can be passed in anywhere that ``put_xxx()`` can passed in. A handler it returned by ``output()``,
     and after being output, the content can also be modified by the handler (See code example below).

    :param contents: The initial contents to be output.
       The item is ``put_xxx()`` call, and any other type will be converted to ``put_text(content)``.
    :return: An OutputHandler instance, the methods of the instance are as follows:

    * ``reset(*contents)`` : Reset original contents to ``contents``
    * ``append(*contents)`` : Append ``contents`` to original contents
    * ``insert(idx, *contents)`` : insert ``contents`` into original contents.

       | when idx>=0, the output content is inserted before the element of the ``idx`` index.
       | when idx<0, the output content is inserted after the element of the ``idx`` index.

    Among them, the parameter ``contents`` is the same as ``output()``.

    :Example:

    .. exportable-codeblock::
        :name: output
        :summary: `output()` usage

        hobby = output(put_text('Coding'))  # equal to output('Coding')
        put_table([
           ['Name', 'Hobbies'],
           ['Wang', hobby]      # hobby is initialized to Coding
        ])
        ## ----

        hobby.reset('Movie')  # hobby is reset to Movie
        ## ----
        hobby.append('Music', put_text('Drama'))  # append Music, Drama to hobby
        ## ----
        hobby.insert(0, put_markdown('**Coding**'))  # insert the Coding into the top of the hobby

    """

    import warnings

```

--------------------------------

### Start PyWebIO Server

Source: https://pywebio.readthedocs.io/en/latest/guide

Starts a PyWebIO web server to host a Python application. The server can be configured with a specific port and debug mode. Debug mode enables automatic reloading on code changes. Remote access can also be enabled for sharing.

```Python
from pywebio import *

def main():  # PyWebIO application function
    name = input.input("what's your name")
    output.put_text("hello", name)

start_server(main, port=8080, debug=True)
```

--------------------------------

### Start Multiple PyWebIO Applications

Source: https://pywebio.readthedocs.io/en/latest/advanced

Demonstrates how to start multiple PyWebIO applications using `start_server()`. Applications can be provided as a list of functions or a dictionary mapping names to functions. Navigation between applications is facilitated using `go_app()` or `put_link()` with the `app` parameter.

```python
from pywebio import put_text, put_buttons, put_link, start_server
from pywebio.session import go_app

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])
```

--------------------------------

### Install PyWebIO Development Version

Source: https://pywebio.readthedocs.io/en/latest/index

Installs the development version of PyWebIO from a GitHub archive. Requires Python 3.5.2 or newer.

```bash
pip3 install -U https://github.com/pywebio/PyWebIO/archive/dev-release.zip
```

--------------------------------

### Start PyWebIO Tornado Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Starts a Tornado server to host PyWebIO applications. This function configures the server with various options like port, host, debugging, CDN usage, static file directories, and remote access.

```Python
def start_server(applications: Union[Callable[[], None], List[Callable[[], None]], Dict[str, Callable[[], None]]],
                 port: int = 0, host: str = '', debug: bool = False, cdn: Union[bool, str] = True,
                 static_dir: Optional[str] = None, remote_access: bool = False, reconnect_timeout: int = 0,
                 allowed_origins: Optional[List[str]] = None, check_origin: Callable[[str], bool] = None,
                 auto_open_webbrowser: bool = False, max_payload_size: Union[int, str] = '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 WebSocket protocol.

    Tornado is the default backend server for PyWebIO applications,
    and ``start_server`` can be imported directly using ``from pywebio import start_server``.

    :param list/dict/callable applications: PyWebIO application.
       Can be a task function, a list of functions, or a dictionary.
       Refer to :ref:`Advanced topic: Multiple applications in start_server() <multiple_app>` for more information.

       When the task function is a coroutine function, use :ref:`Coroutine-based session <coroutine_based_session>` implementation,
       otherwise, use thread-based session implementation.
    :param int port: The port the server listens on.
       When set to ``0``, the server will automatically select a available port.
    :param str host: The host the server listens on. ``host`` may be either an IP address or hostname.
       If it’s a hostname, the server will listen on all IP addresses associated with the name.
       ``host`` may be an empty string or None to listen on all available interfaces.
    :param bool debug: Tornado Server's debug mode. If enabled, the server will automatically reload for code changes.
       See `tornado doc <https://www.tornadoweb.org/en/stable/guide/running.html#debug-mode>`_ for more detail.
    :param bool/str cdn: Whether to load front-end static resources from CDN, the default is ``True``.
       Can also use a string to directly set the url of PyWebIO static resources.
    :param str static_dir: The directory to store the application static files.
       The files in this directory can be accessed via ``http://<host>:<port>/static/files``.
       For example, if there is a ``A/B.jpg`` file in ``static_dir`` path,
       it can be accessed via ``http://<host>:<port>/static/A/B.jpg``.
    :param bool remote_access: Whether to enable remote access, when enabled,
       you can get a temporary public network access address for the current application,

```

--------------------------------

### Start Tornado HTTP Server with PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/platform

Starts a Tornado server to serve PyWebIO applications via HTTP. It allows configuration of session timeouts, cleanup intervals, and maximum payload size.

```python
pywebio.platform.tornado_http.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'_)
```

--------------------------------

### Setup and Run Tornado Server Thread

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Configures and starts a Tornado server in a separate thread for PyWebIO's script mode. It sets logging levels, creates a new asyncio event loop, configures the IOLoop, and starts the server on a specified port. It also spawns callbacks to wait for server shutdown and open a web browser.

```python
    def server_thread():
        from tornado.log import access_log, app_log, gen_log
        access_log.setLevel(logging.ERROR)
        app_log.setLevel(logging.ERROR)
        gen_log.setLevel(logging.ERROR)

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        set_ioloop(tornado.ioloop.IOLoop.current())  

        port = 0
        if os.environ.get("PYWEBIO_SCRIPT_MODE_PORT"):
            port = int(os.environ.get("PYWEBIO_SCRIPT_MODE_PORT"))

        server, port = _setup_server(webio_handler=SingleSessionWSHandler, port=port, host='127.0.0.1',
                                     websocket_max_message_size=parse_file_size('200M'))
        tornado.ioloop.IOLoop.current().spawn_callback(partial(wait_to_stop_loop, server=server))

        if "PYWEBIO_SCRIPT_MODE_PORT" not in os.environ:
            tornado.ioloop.IOLoop.current().spawn_callback(open_webbrowser_on_server_started, '127.0.0.1', port)

        tornado.ioloop.IOLoop.current().start()
        logger.debug('Tornado server exit')

    t = threading.Thread(target=server_thread, name='Tornado-server')
    t.start()

    websocket_conn_opened.wait()
```

--------------------------------

### Start PyWebIO Django Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/django

Starts a Django server to host PyWebIO applications. This function handles server configuration like port, host, static directories, remote access, and integrates with Tornado or Django's built-in runserver.

```Python
def start_server(applications, port=8080, host='', cdn=True,
                 static_dir=None, remote_access=False,
                 allowed_origins=None, check_origin=None,
                 session_expire_seconds=None,
                 session_cleanup_interval=None,
                 debug=False, max_payload_size='200M', **django_options):
    """Start a Django server to provide the PyWebIO application as a web service.

    :param bool debug: Django debug mode.
       See `Django doc <https://docs.djangoproject.com/en/3.0/ref/settings/#debug>`_ for more detail.
    :param django_options: Additional settings to django server.
       For details, please refer: https://docs.djangoproject.com/en/3.0/ref/settings/ .
       Among them, ``DEBUG``, ``ALLOWED_HOSTS``, ``ROOT_URLCONF``, ``SECRET_KEY`` are set by PyWebIO and cannot be specified in ``django_options``.

    The rest arguments of ``start_server()`` have the same meaning as for :func:`pywebio.platform.flask.start_server`
    """
    if port == 0:
        port = get_free_port()

    if not host:
        host = '0.0.0.0'

    max_payload_size = parse_file_size(max_payload_size)
    app = wsgi_app(applications, cdn=cdn, static_dir=static_dir, allowed_origins=allowed_origins,
                   check_origin=check_origin, session_expire_seconds=session_expire_seconds,
                   session_cleanup_interval=session_cleanup_interval,
                   debug=debug, max_payload_size=max_payload_size, **django_options)

    print_listen_address(host, port)

    if remote_access:
        start_remote_access_service(local_port=port)

    use_tornado_wsgi = os.environ.get('PYWEBIO_DJANGO_WITH_TORNADO', True)
    if use_tornado_wsgi:
        import tornado.wsgi
        container = tornado.wsgi.WSGIContainer(app)
        http_server = tornado.httpserver.HTTPServer(container, max_buffer_size=max_payload_size)
        http_server.listen(port, address=host)
        tornado.ioloop.IOLoop.current().start()
    else:
        from django.core.management import call_command
        has_coro_target = any(iscoroutinefunction(target) or isgeneratorfunction(target) for
                              target in make_applications(applications).values())
        if has_coro_target:
            threading.Thread(target=run_event_loop, daemon=True).start()

        call_command('runserver', '%s:%d' % (host, port))
```

--------------------------------

### Put File Example

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/output

Demonstrates how to use the `put_file` function to allow users to download a file. It shows how to read file content and specify the filename and a download prompt.

```Python
content = open('./some-file', 'rb').read()
put_file('hello-world.txt', content, 'download me')
```

--------------------------------

### Start PyWebIO Server with Uvicorn

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/fastapi

Starts a FastAPI/Starlette server using uvicorn to serve PyWebIO applications. It handles server configuration, including host, port, static file serving, and remote access. Additional uvicorn settings can be passed as keyword arguments.

```Python
import asyncio
import uvicorn
from functools import partial
from pywebio.session import Session
from pywebio.utils import get_free_port, parse_file_size
from pywebio.platform.core import OriginChecker
from pywebio.platform.page import page
from pywebio.platform.fastapi import asgi_app, webio_routes, start_remote_access_service, open_webbrowser_on_server_started, print_listen_address, cdn_validation


def start_server(applications, port=0, host='', cdn=True, reconnect_timeout=0,
                 static_dir=None, remote_access=False, debug=False,
                 allowed_origins=None, check_origin=None,
                 auto_open_webbrowser=False,
                 max_payload_size='200M',
                 **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, reconnect_timeout=reconnect_timeout,
                   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('127.0.0.1', port))

    if not host:
        host = '0.0.0.0'

    if port == 0:
        port = get_free_port()

    print_listen_address(host, port)

    if remote_access:
        start_remote_access_service(local_port=port)

    page.MAX_PAYLOAD_SIZE = max_payload_size = parse_file_size(max_payload_size)
    uvicorn_settings = uvicorn_settings or {}
    uvicorn_settings.setdefault('ws_max_size', max_payload_size)

    uvicorn.run(app, host=host, port=port, **uvicorn_settings)

```

--------------------------------

### Example: Displaying Various Loading Indicator Styles in PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/output

This example iterates through different shapes ('border', 'grow') and colors for the loading indicator using `put_loading`, displaying each combination.

```Python
for shape in ('border', 'grow'):
    for color in ('primary', 'secondary', 'success', 'danger', 'warning', 'info', 'light', 'dark'):
        put_text(shape, color)
        put_loading(shape=shape, color=color)
```

--------------------------------

### Start PyWebIO Server with Tornado

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Configures and starts a PyWebIO server using Tornado. It allows customization of port, host, static directories, payload size, and browser auto-opening. It also handles remote access and websocket ping intervals for stable connections.

```python
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)

# covered `os.environ.get()` func with `bool()` to prevent type check error
debug = Session.debug = bool(os.environ.get('PYWEBIO_DEBUG', debug))

# Since some cloud server may close idle connections (such as heroku),
# use `websocket_ping_interval` to  keep the connection alive
tornado_app_settings.setdefault('websocket_ping_interval', 30)
tornado_app_settings.setdefault('websocket_max_message_size', max_payload_size)  # Backward compatible
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, allowed_origins=allowed_origins, check_origin=check_origin,
                        reconnect_timeout=reconnect_timeout)
_, port = _setup_server(webio_handler=handler, port=port, host=host, static_dir=static_dir,
                        max_buffer_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)

if remote_access:
    start_remote_access_service(local_port=port)

tornado.ioloop.IOLoop.current().start()
```

--------------------------------

### PyWebIO Grid Layout Example

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/output

Shows an example of using put_grid to arrange content in a grid format, specifying cell dimensions and layout direction.

```python
put_grid([
    [put_code('A'), put_code('B')],
    [put_code('C'), put_code('D')],
], cell_widths='1fr 2fr', cell_heights='auto auto')
```

--------------------------------

### Install FastAPI/Starlette Dependencies

Source: https://pywebio.readthedocs.io/en/latest/platform

Installs the necessary packages for using PyWebIO with FastAPI or Starlette, including the web frameworks, uvicorn for running the server, and libraries for WebSocket communication.

```Shell
pip3 install -U fastapi starlette uvicorn aiofiles websockets
```

--------------------------------

### Start aiohttp server for PyWebIO applications

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/aiohttp

Starts an aiohttp server to serve PyWebIO applications. It handles application routing, static file serving, websocket settings, and optional features like automatic browser opening and remote access. The function accepts various parameters to configure the server's behavior, including host, port, debug mode, CDN usage, static directory, and origin checking.

```python
def start_server(applications, port=0, host='', debug=False,
                 cdn=True, static_dir=None, remote_access=False,
                 reconnect_timeout=0,
                 allowed_origins=None, check_origin=None,
                 auto_open_webbrowser=False,
                 max_payload_size='200M',
                 websocket_settings=None,
                 **aiohttp_settings):
"""Start a aiohttp server to provide the PyWebIO application as a web service.


    :param dict websocket_settings: The  parameters passed to the constructor of ``aiohttp.web.WebSocketResponse``.
       For details, please refer: https://docs.aiohttp.org/en/stable/web_reference.html#websocketresponse
    :param aiohttp_settings: Additional keyword arguments passed to the constructor of ``aiohttp.web.Application``.
       For details, please refer: https://docs.aiohttp.org/en/stable/web_reference.html#application

    The rest arguments of ``start_server()`` have the same meaning as for :func:`pywebio.platform.tornado.start_server`
    """
    kwargs = locals()

    if not host:
        host = '0.0.0.0'

    if port == 0:
        port = get_free_port()

    cdn = cdn_validation(cdn, 'warn')

    handler = webio_handler(applications, cdn=cdn, allowed_origins=allowed_origins, reconnect_timeout=reconnect_timeout,
                            check_origin=check_origin, max_payload_size=max_payload_size,
                            websocket_settings=websocket_settings)

    app = web.Application(**aiohttp_settings)
    app.router.add_routes([web.get('/', handler)])
    if static_dir is not None:
        app.router.add_routes([web.static('/static', static_dir)])
    app.router.add_routes(static_routes())

    if auto_open_webbrowser:
        asyncio.get_event_loop().create_task(open_webbrowser_on_server_started('127.0.0.1', port))

    debug = Session.debug = os.environ.get('PYWEBIO_DEBUG', debug)
    if debug:
        logging.getLogger("asyncio").setLevel(logging.DEBUG)

    print_listen_address(host, port)

    if remote_access:
        start_remote_access_service(local_port=port)

    web.run_app(app, host=host, port=port)

```

--------------------------------

### Example: Progress Bar Update in PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/output

This example demonstrates how to initialize a progress bar using `put_progressbar` and update its progress iteratively using `set_progressbar` within a loop, including a time delay.

```Python
import time

put_progressbar('bar');
for i in range(1, 11):
    set_progressbar('bar', i / 10)
    time.sleep(0.1)
```

--------------------------------

### Start Tornado Server for PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/platform

Starts a Tornado server to serve PyWebIO applications via WebSocket. It supports various configurations for applications, ports, hosts, debugging, CDN usage, static file directories, remote access, reconnection timeouts, allowed origins, and payload sizes. The server uses Tornado's default backend for PyWebIO.

```python
from pywebio import start_server

# Example usage:
# start_server(my_pywebio_app, port=8080, cdn=False, static_dir='./static')
```

--------------------------------

### Start PyWebIO Flask Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/flask

Starts a Flask development server to host PyWebIO applications. This function allows configuration of the server port, host, CDN usage, static file directory, remote access, session management, and debugging.

```Python
def start_server(applications, port=8080, host='', cdn=True,
                 static_dir=None, remote_access=False,
                 allowed_origins=None, check_origin=None,
                 session_expire_seconds=None,
                 session_cleanup_interval=None,
                 debug=False,
                 max_payload_size='200M',
                 **flask_options):
    cdn = cdn_validation(cdn, 'error')

    if remote_access:
        start_remote_access_service(port=port, host=host)

    app = wsgi_app(applications, cdn=cdn, static_dir=static_dir,
                   allowed_origins=allowed_origins, check_origin=check_origin,
                   session_expire_seconds=session_expire_seconds,
                   session_cleanup_interval=session_cleanup_interval,
                   max_payload_size=max_payload_size)

    # Flask's default server is not suitable for production
    # For production, use a production WSGI server like Gunicorn or uWSGI
    app.run(host=host, port=port, debug=debug, **flask_options)

    print_listen_address(port, host)

    # If remote_access is enabled, the remote access service will run in a separate thread
    # The main thread will exit after the Flask server stops.
    # If you need the remote access service to keep running, you might need to manage threads explicitly.
    if remote_access:
        # Example: Keep the main thread alive if remote access needs to persist
        # This is a basic example and might need more robust thread management
        pass

```

--------------------------------

### Install Django for PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/platform

Installs Django version 2.2 or higher, which is required when using Django as a PyWebIO backend server.

```bash
pip3 install -U django>=2.2
```

--------------------------------

### Install Flask for PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/platform

Command to install or upgrade Flask to version 0.10 or higher, a prerequisite for using PyWebIO with Flask.

```bash
pip3 install -U flask>=0.10
```

--------------------------------

### Start PyWebIO Server with FastAPI/Starlette and Uvicorn

Source: https://pywebio.readthedocs.io/en/latest/platform

Starts a FastAPI/Starlette server using uvicorn to serve PyWebIO applications. It allows configuration of the port, host, CDN usage, reconnection timeouts, static directories, remote access, debug mode, allowed origins, and browser auto-opening. Additional uvicorn settings can also be passed.

```Python
pywebio.platform.fastapi.start_server(_applications_ , _port =0_, _host =''_, _cdn =True_, _reconnect_timeout =0_, _static_dir =None_, _remote_access =False_, _debug =False_, _allowed_origins =None_, _check_origin =None_, _auto_open_webbrowser =False_, _max_payload_size ='200M'_, _** uvicorn_settings_)
```

--------------------------------

### Open Webbrowser on Server Start

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Attempts to open the PyWebIO application in a web browser when the Tornado server starts. It waits for the server port to be available before attempting to open the URL.

```Python
async def open_webbrowser_on_server_started(host, port):
    url = 'http://%s:%s' % (host, port)
    is_open = await wait_host_port(host, port, duration=20)
    if is_open:
        logger.info('Try open %s in web browser' % url)
        # webbrowser.open() may block, so invoke it in thread
        threading.Thread(target=webbrowser.open, args=(url,), daemon=True).start()
    else:
        logger.error('Open %s in web browser failed.' % url)
```

--------------------------------

### File Upload Example

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/input

Provides an example of using the `file_upload()` function in PyWebIO to allow users to upload a file. The uploaded file's content can then be processed, such as saving it to the server.

```python
f = input.file_upload("Upload a file")
open('asset/'+f['filename'], 'wb').write(f['content'])
```

--------------------------------

### PyWebIO Platform Integration

Source: https://pywebio.readthedocs.io/en/latest/genindex

This snippet lists functions for integrating PyWebIO with different web frameworks and platforms, such as starting servers and handling HTTP requests.

```python
from pywebio.platform import path_deploy, path_deploy_http
from pywebio.platform.aiohttp import start_server as start_server_aiohttp
from pywebio.platform.django import start_server as start_server_django
from pywebio.platform.fastapi import start_server as start_server_fastapi
from pywebio.platform.flask import start_server as start_server_flask, FlaskHttpContext
from pywebio.platform.tornado import start_server as start_server_tornado
from pywebio.platform.tornado_http import start_server as start_server_tornado_http
from pywebio.platform.flask import request_body, request_headers, request_method, request_obj, request_url_parameter, set_content, set_header, set_status
from pywebio.platform import seo
```

--------------------------------

### Example: Context Manager for Loading Indicator in PyWebIO

Source: https://pywebio.readthedocs.io/en/latest/output

This example shows how to use `put_loading` as a context manager. The loading prompt appears when entering the `with` block and disappears automatically when the block finishes, useful for time-consuming operations.

```Python
with put_loading():
    put_text("Start waiting...")
    time.sleep(3)  # Some time-consuming operations
put_text("The answer of the universe is 42")
```

--------------------------------

### Nginx: WebSocket Configuration Example

Source: https://pywebio.readthedocs.io/en/latest/advanced

Provides an example Nginx configuration snippet to enable WebSocket protocol support. This is crucial for PyWebIO applications running with frameworks like aiohttp or FastAPI/Starlette when deployed behind a reverse proxy.

```Nginx
location /tool {
    proxy_pass http://localhost:8000;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
}
```

--------------------------------

### Start PyWebIO Server with Flask

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/flask

This function starts a PyWebIO server using Flask. It handles session management, debug mode, and integrates with Flask's run method. It also includes logic for remote access and running coroutines.

```Python
def start_server(applications, host=None, port=0, cdn=None, static_dir=None, allowed_origins=None, check_origin=None, session_expire_seconds=300, session_cleanup_interval=300, debug=False, max_payload_size=1024*1024, flask_options=None, remote_access=False):
    """Start PyWebIO server with Flask.

    :param applications: A dict of application name to application coroutine function or generator function.
    :param host: Host address to bind to. Default is '0.0.0.0'.
    :param port: Port to bind to. Default is 0, which means a free port will be used.
    :param cdn: CDN URL for PyWebIO resources. If None, use local resources.
    :param static_dir: Directory for static files. If None, use PyWebIO's default static files.
    :param allowed_origins: List of allowed origins for WebSocket connections.
    :param check_origin: Callback function to check WebSocket origin.
    :param session_expire_seconds: Session expiration time in seconds. Default is 300s.
       If no client message is received within ``session_expire_seconds``, the session will be considered expired.
    :param session_cleanup_interval: Session cleanup interval, in seconds(default 300s).
       The server will periodically clean up expired sessions and release the resources occupied by the sessions.
    :param debug: Flask debug mode. 
       If enabled, the server will automatically reload for code changes.
    :param max_payload_size: Max size of a request body which Flask can accept.
    :param flask_options: Additional keyword arguments passed to the ``flask.Flask.run``.
       For details, please refer: https://flask.palletsprojects.com/en/1.1.x/api/#flask.Flask.run

    The arguments of ``start_server()`` have the same meaning as for :func:`pywebio.platform.tornado.start_server`
    """
    if not host:
        host = '0.0.0.0'

    if port == 0:
        port = get_free_port()

    app = wsgi_app(applications, cdn=cdn, static_dir=static_dir, allowed_origins=allowed_origins,
                   check_origin=check_origin, session_expire_seconds=session_expire_seconds,
                   session_cleanup_interval=session_cleanup_interval, max_payload_size=max_payload_size)

    print_listen_address(host, port)

    debug = Session.debug = os.environ.get('PYWEBIO_DEBUG', debug)
    if not debug:
        logging.getLogger('werkzeug').setLevel(logging.WARNING)

    running_from_reloader = werkzeug.serving.is_running_from_reloader()
    if remote_access and not running_from_reloader:
        start_remote_access_service(local_port=port)

    has_coro_target = any(iscoroutinefunction(target) or isgeneratorfunction(target) for
                          target in make_applications(applications).values())
    if has_coro_target and not running_from_reloader:
        threading.Thread(target=run_event_loop, daemon=True).start()

    app.run(host=host, port=port, debug=debug, threaded=True, use_evalex=False, **flask_options)

```

--------------------------------

### Start PyWebIO Script Mode Server

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado

Starts a server for PyWebIO's script mode, listening on an available port and automatically opening a web browser. It supports setting the port via the PYWEBIO_SCRIPT_MODE_PORT environment variable and disabling the auto-open feature for testing purposes.

```python
def start_server_in_current_thread_session():
    """启动 script mode 的server，监听可用端口，并自动打开浏览器
    Start the server for script mode, and automatically open the browser when the server port is available.

    PYWEBIO_SCRIPT_MODE_PORT环境变量可以设置监听端口，并关闭自动打开浏览器，用于测试
    The PYWEBIO_SCRIPT_MODE_PORT environment variable can set the listening port, just used in testing.
    """
    websocket_conn_opened = threading.Event()
    thread = threading.current_thread()

    class SingleSessionWSHandler(_webio_handler(cdn=False)):
        session: ScriptModeSession = None
        instance: typing.ClassVar = None
        closed = False

        def send_msg_to_client(self, session):
            for msg in session.get_task_commands():
                try:
                    self.write_message(json.dumps(msg))
                except TypeError as e:
                    logger.exception('Data serialization error: %s\n' \
                                     'This may be because you pass the wrong type of parameter to the function' \
                                     ' of PyWebIO.\nData content: %s', e, msg)

        def open(self):
            if SingleSessionWSHandler.session is None:
                SingleSessionWSHandler.instance = self
```

--------------------------------

### PyWebIO Input Group Example

Source: https://pywebio.readthedocs.io/en/latest/guide

Demonstrates how to use `input_group` to collect multiple user inputs in a single form. It shows how to define individual inputs with names and types, and how to access the collected data as a dictionary. Includes an example of a custom validation function for the entire form.

```python
from pywebio import input, put_text, output
from pywebio.input import input_group, NUMBER

def check_age(age):
    if age <= 0:
        return 'Age must be positive'

data = input_group("Basic info", [
  input('Input your name', name='name'),
  input('Input your age', name='age', type=NUMBER, validate=check_age)
])
put_text(data['name'], data['age'])
```

```python
def check_form(data):
    if len(data['name']) > 6:
        return ('name', 'Name too long!')
    if data['age'] <= 0:
        return ('age', 'Age can not be negative!')
```

--------------------------------

### File Picker Widget Example

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio_battery/file_picker

Demonstrates how to use the file_picker function to select Python files from the local file system. The selected files are then displayed using put_text.

```python
files = file_picker('.', multiple=True, accept='py')
put_text(files)
```

--------------------------------

### Start PyWebIO Aiohttp Server

Source: https://pywebio.readthedocs.io/en/latest/platform

Initiates an aiohttp server to serve PyWebIO applications. Allows configuration of port, host, debug mode, CDN usage, static directory, and WebSocket settings. Additional aiohttp application settings can be provided via aiohttp_settings.

```python
pywebio.platform.aiohttp.start_server(_applications_ , _port =0_, _host =''_, _debug =False_, _cdn =True_, _static_dir =None_, _remote_access =False_, _reconnect_timeout =0_, _allowed_origins =None_, _check_origin =None_, _auto_open_webbrowser =False_, _max_payload_size ='200M'_, _websocket_settings =None_, _** aiohttp_settings_)
```

--------------------------------

### Get Current Session Implementation

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/session

Retrieves the current session implementation. If no session is active, it defaults to `ScriptModeSession` and starts a server. It handles cases with a single active implementation or multiple, attempting to find the active one.

```python
def get_session_implement():
"""获取当前会话实现。仅供内部实现使用。应在会话上下文中调用
    Get the current session implementation. For internal implementation use only. Should be called in session context"""
    if not _active_session_cls:
        _active_session_cls.append(ScriptModeSession)
        _start_script_mode_server()

    # 当前正在使用的会话实现只有一个
    # There is only one session implementation currently in use
    if len(_active_session_cls) == 1:
        return _active_session_cls[0]

    # 当前有多个正在使用的会话实现
    # There are currently multiple session implementations in use
    for cls in _active_session_cls:
        try:
            cls.get_current_session()
            return cls
        except SessionNotFoundException:
            pass

    raise SessionNotFoundException
```

--------------------------------

### Get Pin Widget Value

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/pin

Shows how to retrieve the value of a pin widget using the `pin` object. After creating an input widget named 'pin_name', this example demonstrates how to display its current value when a button is clicked.

```python
from pywebio.pin import put_input, pin
from pywebio.output import put_text, put_buttons

put_input('pin_name')
put_buttons(['Get Pin Value'], lambda _: put_text(pin.pin_name))
```

--------------------------------

### PyWebIO Server Mode - Invalid Code Example

Source: https://pywebio.readthedocs.io/en/latest/guide

Illustrates an invalid code pattern in PyWebIO's server mode where interactive functions are called outside of a PyWebIO application context. This will lead to an error.

```Python
import pywebio
from pywebio.input import input

port = input('Input port number:')   # ❌ error
pywebio.start_server(my_task_func, port=int(port))
```

--------------------------------

### start_server for PyWebIO with Tornado

Source: https://pywebio.readthedocs.io/en/latest/_modules/pywebio/platform/tornado_http

Starts a Tornado web server to host PyWebIO applications. It configures the server with various options including host, port, debugging, CDN usage, static file directory, session management, and maximum payload size.

```Python
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):

    if not host:
        host = '0.0.0.0'

    cdn = cdn_validation(cdn, 'warn')

    set_ioloop(tornado.ioloop.IOLoop.current())

    cdn = cdn_validation(cdn, 'warn')

    page.MAX_PAYLOAD_SIZE = max_payload_size = parse_file_size(max_payload_size)

    debug = Session.debug = os.environ.get('PYWEBIO_DEBUG', debug)

```

--------------------------------

### Concurrency with run_async in PyWebIO Coroutine Session

Source: https://pywebio.readthedocs.io/en/latest/advanced

Illustrates how to run coroutines concurrently within a PyWebIO coroutine session using `run_async`. This example starts a counter coroutine that prints numbers with a delay. Requires `pywebio.start_server` and `pywebio.session.run_async`.

```python
from pywebio import start_server
from pywebio.session import run_async
import asyncio

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

--------------------------------

### PyWebIO Basic Text Input

Source: https://pywebio.readthedocs.io/en/latest/guide

Demonstrates how to get basic text input from a user in a web browser using PyWebIO's `input` function. The function is blocking and returns the user's input after submission.

```Python
from pywebio.input import input
from pywebio.input import NUMBER

age = input("How old are you?", type=NUMBER)
```

--------------------------------

### PyWebIO: Nested Layout with Rows and Columns

Source: https://pywebio.readthedocs.io/en/latest/guide

Shows how to create complex layouts by nesting `put_row` and `put_column` functions. This example demonstrates arranging code blocks and other elements in a structured manner, including using `None` for spacing.

```python
put_row([
    put_column([
        put_code('A'),
        put_row([
            put_code('B1'), None,
            put_code('B2'), None,
            put_code('B3'),
        ]),
        put_code('C'),
    ]), None,
    put_code('D'), None,
    put_code('E')
])
```

--------------------------------

### Start PyWebIO Django Server

Source: https://pywebio.readthedocs.io/en/latest/platform

Launches a Django server to host PyWebIO applications as a web service. Supports various configurations including port, host, CDN usage, static file directory, and debug mode. Additional Django settings can be passed via django_options.

```python
pywebio.platform.django.start_server(_applications_ , _port =8080_, _host =''_, _cdn =True_, _static_dir =None_, _remote_access =False_, _allowed_origins =None_, _check_origin =None_, _session_expire_seconds =None_, _session_cleanup_interval =None_, _debug =False_, _max_payload_size ='200M'_, _** django_options_)
```

--------------------------------

### PyWebIO: Clear, Remove, and Scroll Scopes

Source: https://pywebio.readthedocs.io/en/latest/guide

Provides examples of PyWebIO scope control functions: `clear(scope)` to remove content from a scope, `remove(scope)` to delete the scope entirely, and `scroll_to(scope)` to bring a scope into view.

```python
# Clear the contents of the scope
clear(scope)

# Remove scope
remove(scope)

# Scroll the page to the scope
scroll_to(scope)
```
Tokens:
Raw