nicegui/nicegui/ui_run.py

import multiprocessing
import os
import sys
from pathlib import Path
from typing import Any, List, Literal, Optional, Tuple, TypedDict, Union

from fastapi.middleware.gzip import GZipMiddleware
from starlette.routing import Route
from uvicorn.main import STARTUP_FAILURE
from uvicorn.supervisors import ChangeReload, Multiprocess

import __main__

from . import core, helpers
from . import native as native_module
from .air import Air
from .client import Client
from .language import Language
from .logging import log
from .middlewares import RedirectWithPrefixMiddleware
from .server import CustomServerConfig, Server

APP_IMPORT_STRING = 'nicegui:app'


class ContactDict(TypedDict):
    name: Optional[str]
    url: Optional[str]
    email: Optional[str]


class LicenseInfoDict(TypedDict):
    name: str
    identifier: Optional[str]
    url: Optional[str]


class DocsConfig(TypedDict):
    title: Optional[str]
    summary: Optional[str]
    description: Optional[str]
    version: Optional[str]
    terms_of_service: Optional[str]
    contact: Optional[ContactDict]
    license_info: Optional[LicenseInfoDict]


def run(*,
        host: Optional[str] = None,
        port: Optional[int] = None,
        title: str = 'NiceGUI',
        viewport: str = 'width=device-width, initial-scale=1',
        favicon: Optional[Union[str, Path]] = None,
        dark: Optional[bool] = False,
        language: Language = 'en-US',
        binding_refresh_interval: float = 0.1,
        reconnect_timeout: float = 3.0,
        message_history_length: int = 1000,
        fastapi_docs: Union[bool, DocsConfig] = False,
        show: bool = True,
        on_air: Optional[Union[str, Literal[True]]] = None,
        native: bool = False,
        window_size: Optional[Tuple[int, int]] = None,
        fullscreen: bool = False,
        frameless: bool = False,
        reload: bool = True,
        uvicorn_logging_level: str = 'warning',
        uvicorn_reload_dirs: str = '.',
        uvicorn_reload_includes: str = '*.py',
        uvicorn_reload_excludes: str = '.*, .py[cod], .sw.*, ~*',
        tailwind: bool = True,
        prod_js: bool = True,
        endpoint_documentation: Literal['none', 'internal', 'page', 'all'] = 'none',
        storage_secret: Optional[str] = None,
        show_welcome_message: bool = True,
        **kwargs: Any,
        ) -> None:
    """ui.run

    You can call `ui.run()` with optional arguments.
    Most of them only apply after stopping and fully restarting the app and do not apply with auto-reloading.

    :param host: start server with this host (defaults to `'127.0.0.1` in native mode, otherwise `'0.0.0.0'`)
    :param port: use this port (default: 8080 in normal mode, and an automatically determined open port in native mode)
    :param title: page title (default: `'NiceGUI'`, can be overwritten per page)
    :param viewport: page meta viewport content (default: `'width=device-width, initial-scale=1'`, can be overwritten per page)
    :param favicon: relative filepath, absolute URL to a favicon (default: `None`, NiceGUI icon will be used) or emoji (e.g. `'🚀'`, works for most browsers)
    :param dark: whether to use Quasar's dark mode (default: `False`, use `None` for "auto" mode)
    :param language: language for Quasar elements (default: `'en-US'`)
    :param binding_refresh_interval: time between binding updates (default: `0.1` seconds, bigger is more CPU friendly)
    :param reconnect_timeout: maximum time the server waits for the browser to reconnect (default: 3.0 seconds)
    :param message_history_length: maximum number of messages that will be stored and resent after a connection interruption (default: 1000, use 0 to disable)
    :param fastapi_docs: enable FastAPI's automatic documentation with Swagger UI, ReDoc, and OpenAPI JSON (bool or dictionary as described `here<https://fastapi.tiangolo.com/tutorial/metadata/>`_, default: `False`)
    :param show: automatically open the UI in a browser tab (default: `True`)
    :param on_air: tech preview: `allows temporary remote access <https://nicegui.io/documentation/section_configuration_deployment#nicegui_on_air>`_ if set to `True` (default: disabled)
    :param native: open the UI in a native window of size 800x600 (default: `False`, deactivates `show`, automatically finds an open port)
    :param window_size: open the UI in a native window with the provided size (e.g. `(1024, 786)`, default: `None`, also activates `native`)
    :param fullscreen: open the UI in a fullscreen window (default: `False`, also activates `native`)
    :param frameless: open the UI in a frameless window (default: `False`, also activates `native`)
    :param reload: automatically reload the UI on file changes (default: `True`)
    :param uvicorn_logging_level: logging level for uvicorn server (default: `'warning'`)
    :param uvicorn_reload_dirs: string with comma-separated list for directories to be monitored (default is current working directory only)
    :param uvicorn_reload_includes: string with comma-separated list of glob-patterns which trigger reload on modification (default: `'*.py'`)
    :param uvicorn_reload_excludes: string with comma-separated list of glob-patterns which should be ignored for reload (default: `'.*, .py[cod], .sw.*, ~*'`)
    :param tailwind: whether to use Tailwind (experimental, default: `True`)
    :param prod_js: whether to use the production version of Vue and Quasar dependencies (default: `True`)
    :param endpoint_documentation: control what endpoints appear in the autogenerated OpenAPI docs (default: 'none', options: 'none', 'internal', 'page', 'all')
    :param storage_secret: secret key for browser-based storage (default: `None`, a value is required to enable ui.storage.individual and ui.storage.browser)
    :param show_welcome_message: whether to show the welcome message (default: `True`)
    :param kwargs: additional keyword arguments are passed to `uvicorn.run`
    """
    core.app.config.add_run_config(
        reload=reload,
        title=title,
        viewport=viewport,
        favicon=favicon,
        dark=dark,
        language=language,
        binding_refresh_interval=binding_refresh_interval,
        reconnect_timeout=reconnect_timeout,
        message_history_length=message_history_length,
        tailwind=tailwind,
        prod_js=prod_js,
        show_welcome_message=show_welcome_message,
    )
    core.app.config.endpoint_documentation = endpoint_documentation
    if not helpers.is_pytest():
        core.app.add_middleware(GZipMiddleware)
    core.app.add_middleware(RedirectWithPrefixMiddleware)

    for route in core.app.routes:
        if not isinstance(route, Route):
            continue
        if route.path.startswith('/_nicegui') and hasattr(route, 'methods'):
            route.include_in_schema = endpoint_documentation in {'internal', 'all'}
        if route.path == '/' or route.path in Client.page_routes.values():
            route.include_in_schema = endpoint_documentation in {'page', 'all'}

    if fastapi_docs:
        if not core.app.docs_url:
            core.app.docs_url = '/docs'
        if not core.app.redoc_url:
            core.app.redoc_url = '/redoc'
        if not core.app.openapi_url:
            core.app.openapi_url = '/openapi.json'
        if isinstance(fastapi_docs, dict):
            core.app.title = fastapi_docs.get('title') or title
            core.app.summary = fastapi_docs.get('summary')
            core.app.description = fastapi_docs.get('description') or ''
            core.app.version = fastapi_docs.get('version') or '0.1.0'
            core.app.terms_of_service = fastapi_docs.get('terms_of_service')
            contact = fastapi_docs.get('contact')
            license_info = fastapi_docs.get('license_info')
            core.app.contact = dict(contact) if contact else None
            core.app.license_info = dict(license_info) if license_info else None
        core.app.setup()

    if on_air:
        core.air = Air('' if on_air is True else on_air)

    if multiprocessing.current_process().name != 'MainProcess':
        return

    if reload and not hasattr(__main__, '__file__'):
        log.warning('disabling auto-reloading because is is only supported when running from a file')
        core.app.config.reload = reload = False

    if fullscreen:
        native = True
    if frameless:
        native = True
    if window_size:
        native = True
    if native:
        show = False
        host = host or '127.0.0.1'
        port = port or native_module.find_open_port()
        width, height = window_size or (800, 600)
        native_module.activate(host, port, title, width, height, fullscreen, frameless)
    else:
        port = port or 8080
        host = host or '0.0.0.0'
    assert host is not None
    assert port is not None

    # NOTE: We save host and port in environment variables so the subprocess started in reload mode can access them.
    os.environ['NICEGUI_HOST'] = host
    os.environ['NICEGUI_PORT'] = str(port)

    if show:
        helpers.schedule_browser(host, port)

    def split_args(args: str) -> List[str]:
        return [a.strip() for a in args.split(',')]

    if kwargs.get('workers', 1) > 1:
        raise ValueError('NiceGUI does not support multiple workers yet.')

    # NOTE: The following lines are basically a copy of `uvicorn.run`, but keep a reference to the `server`.

    config = CustomServerConfig(
        APP_IMPORT_STRING if reload else core.app,
        host=host,
        port=port,
        reload=reload,
        reload_includes=split_args(uvicorn_reload_includes) if reload else None,
        reload_excludes=split_args(uvicorn_reload_excludes) if reload else None,
        reload_dirs=split_args(uvicorn_reload_dirs) if reload else None,
        log_level=uvicorn_logging_level,
        **kwargs,
    )
    config.storage_secret = storage_secret
    config.method_queue = native_module.method_queue if native else None
    config.response_queue = native_module.response_queue if native else None
    Server.create_singleton(config)

    if (reload or config.workers > 1) and not isinstance(config.app, str):
        log.warning('You must pass the application as an import string to enable "reload" or "workers".')
        sys.exit(1)

    if config.should_reload:
        sock = config.bind_socket()
        ChangeReload(config, target=Server.instance.run, sockets=[sock]).run()
    elif config.workers > 1:
        sock = config.bind_socket()
        Multiprocess(config, target=Server.instance.run, sockets=[sock]).run()
    else:
        Server.instance.run()
    if config.uds:
        os.remove(config.uds)  # pragma: py-win32

    if not Server.instance.started and not config.should_reload and config.workers == 1:
        sys.exit(STARTUP_FAILURE)