Help for web.server

Project

web.app

Function

setup

Sample CLI

gway web.app setup

References

['debug', 'info', 'resource', 'to_html', 'to_list', 'unwrap', 'version', 'web', 'web.app_url', 'web.redirect_error', 'web.static_url', 'web.work_url']

Full Code

def setup(*,
    app=None,
    project="web.site",
    path=None,
    static="static",
    work="work",
    home: str = "readme",
    prefix: str = "view",
    navbar: bool = True,
):    
    """
    Configure one or more Bottle-based apps. Use web server start-app to launch.
    """
    global _version
    _version = _version or gw.version() 

    # Normalize `project` into a list of project names
    projects = gw.to_list(project, flat=True)

    # Determine default `path` if not provided
    if path is None:
        first_proj = projects[0]
        path = "gway" if first_proj == "web.site" else first_proj.replace(".", "/")

    _is_new_app = not (app := gw.unwrap(app, Bottle) if (oapp := app) else None)
    gw.debug(f"Unwrapped {app=} from {oapp=} ({_is_new_app=})")

    if _is_new_app:
        gw.info("No Bottle app found; creating a new Bottle app.")
        app = Bottle()

        # Define URL-building helpers
        gw.web.static_url = lambda *args, **kwargs: build_url(static, *args, **kwargs)
        gw.web.work_url = lambda *args, **kwargs: build_url(work, *args, **kwargs)
        gw.web.app_url = lambda *args, **kwargs: build_url(path, *args, **kwargs)
        gw.web.redirect_error = redirect_error

    @app.route("/accept-cookies", method="POST")
    def accept_cookies():
        response.set_cookie("cookies_accepted", "yes")
        redirect_url = request.forms.get("next", "/readme")
        response.status = 303
        if not redirect_url.startswith("/"):
            redirect_url = f"/{redirect_url}"
        response.set_header("Location", f"/{path}{redirect_url}")
        return ""

    if static:
        @app.route(f"/{static}/<filename:path>")
        def send_static(filename):
            return static_file(filename, root=gw.resource("data", "static"))

    if work:
        @app.route(f"/{work}/<filename:path>")
        def send_work(filename):
            filename = filename.replace('-', '_')
            return static_file(filename, root=gw.resource("work", "shared"))
        
    # TODO: When passing a param to a view, such as from a form that was left empty by the user
    # avoid passing None, instead pass "" (empty string) as it may differentiate between a 
    # form that has not been submitted and one submitted with empty fields.

    @app.route(f"/{path}/<view:path>", method=["GET", "POST"])
    def view_dispatch(view):
        nonlocal navbar
        segments = [s for s in view.strip("/").split("/") if s]
        if not segments:
            segments = [home]
        view_name = segments[0].replace("-", "_")
        args = segments[1:]
        kwargs = dict(request.query)
        if request.method == "POST":
            try:
                if request.json:
                    kwargs.update(request.json)
                elif request.forms:
                    kwargs.update(request.forms.decode())
            except Exception as e:
                return redirect_error(e, note="Error loading JSON payload", broken_view_name=view_name)

        sources = []
        for proj_name in projects:
            try:
                sources.append(gw[proj_name])
            except Exception:
                continue

        for source in sources:
            view_func = getattr(source, f"{prefix}_{view_name}", None)
            if callable(view_func):
                break
        else:
            return redirect_error(
                note=f"View '{prefix}_{view_name}' not found in any project: {projects}",
                broken_view_name=view_name,
                default=f"/{path}/{home}" if path and home else "/gway/readme"
            )

        try:
            gw.debug(f"Dispatch to {view_func.__name__} (args={args}, kwargs={kwargs})")
            content = view_func(*args, **kwargs)
            if content and not isinstance(content, str):
                content = gw.to_html(content)
            visited = update_visited(view_name)
        except HTTPResponse as resp:
            return resp
        except Exception as e:
            return redirect_error(e, note="Error during view execution", broken_view_name=view_name)

        full_url = request.fullpath
        if request.query_string:
            full_url += "?" + request.query_string
        if navbar is True:
            navbar = render_navbar(visited, path, current_url=full_url)
        if not cookies_enabled():
            consent_box = f"""
                <div class="consent-box">
                <form action="/accept-cookies" method="post">
                    <input type="hidden" name="next" value="/{view}" />
                    This site uses cookies to improve your experience.
                    <button type="submit">Accept</button>
                </form>
                </div>
            """
            content = consent_box + content

        style_param = request.query.get("css") or request.query.get("style")
        if style_param:
            if not style_param.endswith(".css"):
                style_param += ".css"
            response.set_cookie("css", style_param, path="/")
            css_files = ["default.css", style_param]
        else:
            css_cookie = request.get_cookie("css", "")
            css_files = ["default.css"] + [c.strip() for c in css_cookie.split(",") if c.strip()]

        return render_template(
            title="GWAY - " + view_name.replace("_", " ").title(),
            navbar=navbar,
            content=content,
            static=static,
            css_files=css_files
        )

    @app.route("/", method=["GET", "POST"])
    def index():
        response.status = 302
        response.set_header("Location", f"/{path}/readme")
        return ""
    
    @app.error(404)
    def handle_404(error):
        fallback = "/gway/readme"
        try:
            return redirect_error(
                error,
                note=f"404 Not Found: {request.url}",
                default=f"/{path}/{home}" if path and home else fallback
            )
        except Exception as e:
            return redirect_error(e, note="Failed during 404 fallback", default=fallback)

    if _is_new_app:
        app = security_middleware(app)
        return app if not oapp else (oapp, app)
    
    return oapp

Project

web.proxy

Function

setup_app

Sample CLI

gway web.proxy setup-app

References

['unwrap']

Full Code

def setup_app(*, endpoint: str, app=None, websockets: bool = False, path: str = "/"):
    """
    Create an HTTP (and optional WebSocket) proxy to the given endpoint.
    Accepts positional apps, the `app=` kwarg, or both. Flattens any iterables
    and selects apps by type using gw.filter_apps.

    Returns a single app if one is provided, otherwise a tuple of apps.
    """
    # selectors for app types
    from bottle import Bottle

    def is_bottle_app(candidate) -> bool:
        return isinstance(candidate, Bottle)

    def is_fastapi_app(candidate) -> bool:
        return hasattr(candidate, "websocket")

    # collect apps by type
    bottle_app = gw.unwrap(app, Bottle)
    fastapi_app = gw.unwrap(app, FastAPI)

    prepared = []

    # if no matching apps, default to a new Bottle
    if not bottle_app and not fastapi_app:
        default = Bottle()
        prepared.append(_wire_proxy(default, endpoint, websockets, path))
    elif bottle_app:
        prepared.append(_wire_proxy(bottle_app, endpoint, websockets, path))
    elif fastapi_app:
        prepared.append(_wire_proxy(fastapi_app, endpoint, websockets, path))

    # TODO: Test if this is properly compatible with web.server.start_app
    return prepared[0] if len(prepared) == 1 else tuple(prepared)

Project

web.server

Function

iterable

Sample CLI

gway web.server iterable

Full Code

@set_module('numpy')
def iterable(y):
    """
    Check whether or not an object can be iterated over.

    Parameters
    ----------
    y : object
      Input object.

    Returns
    -------
    b : bool
      Return ``True`` if the object has an iterator method or is a
      sequence and ``False`` otherwise.


    Examples
    --------
    >>> import numpy as np
    >>> np.iterable([1, 2, 3])
    True
    >>> np.iterable(2)
    False

    Notes
    -----
    In most cases, the results of ``np.iterable(obj)`` are consistent with
    ``isinstance(obj, collections.abc.Iterable)``. One notable exception is
    the treatment of 0-dimensional arrays::

        >>> from collections.abc import Iterable
        >>> a = np.array(1.0)  # 0-dimensional numpy array
        >>> isinstance(a, Iterable)
        True
        >>> np.iterable(a)
        False

    """
    try:
        iter(y)
    except TypeError:
        return False
    return True

Project

web.server

Function

start_app

Sample CLI

gway web.server start-app

References

['info', 'warning', 'web', 'web.app', 'web.app.setup', 'web.server', 'web.server.start']

Full Code

def start_app(*,
    host="[WEBSITE_HOST|127.0.0.1]",
    port="[WEBSITE_PORT|8888]",
    debug=False,
    proxy=None,
    app=None,
    daemon=False,
    threaded=True,
    is_worker=False,
):
    """Start an HTTP (WSGI) or ASGI server to host the given application.

    - If `app` is a FastAPI instance, runs with Uvicorn.
    - If `app` is a WSGI app (Bottle, Paste URLMap, or generic WSGI callables), uses Paste+ws4py or Bottle.
    - If `app` is a zero-arg factory, it will be invoked (supporting sync or async factories).
    - If `app` is a list of apps, each will be run in its own thread (each on an incremented port).
    """
    import inspect
    import asyncio

    def run_server():
        nonlocal app
        all_apps = app if iterable(app) else (app, )

        # B. Dispatch multiple apps in threads if we aren't already in a worker
        if not is_worker and len(all_apps) > 1:
            from threading import Thread
            from collections import Counter
            threads = []
            app_types = []
            gw.info(f"Starting {len(all_apps)} apps in parallel threads.")
            for i, sub_app in enumerate(all_apps):
                try:
                    from fastapi import FastAPI
                    app_type = "FastAPI" if isinstance(sub_app, FastAPI) else type(sub_app).__name__
                except ImportError:
                    app_type = type(sub_app).__name__
                port_i = int(port) + i
                gw.info(f"  App {i+1}: type={app_type}, port={port_i}")
                app_types.append(app_type)

                t = Thread(
                    target=gw.web.server.start,
                    kwargs=dict(
                        host=host,
                        port=port_i,
                        debug=debug,
                        proxy=proxy,
                        app=sub_app,
                        daemon=daemon,
                        threaded=threaded,
                        is_worker=True,
                    ),
                    daemon=daemon,
                )
                t.start()
                threads.append(t)

            type_summary = Counter(app_types)
            summary_str = ", ".join(f"{count}×{t}" for t, count in type_summary.items())
            gw.info(f"All {len(all_apps)} apps started. Types: {summary_str}")

            if not daemon:
                for t in threads:
                    t.join()
            return

        # 1. If no apps passed, fallback to default app
        if not all_apps:
            gw.warning(
                "Building default app (app is None). Run with --app default to silence.")
            app = gw.web.app.setup(app=None)
        else:
            app = all_apps[0]  # Run the first (or only) app normally

        # 2. Wrap with proxy if requested
        if proxy:
            from .proxy import setup_app as setup_proxy
            app = setup_proxy(endpoint=proxy, app=app)

        # 3. If app is a zero-arg factory, invoke it
        if callable(app):
            sig = inspect.signature(app)
            if len(sig.parameters) == 0:
                gw.info(f"Calling app factory: {app}")
                maybe_app = app()
                if inspect.isawaitable(maybe_app):
                    maybe_app = asyncio.get_event_loop().run_until_complete(maybe_app)
                app = maybe_app
            else:
                gw.info(f"Detected callable WSGI/ASGI app: {app}")

        gw.info(f"Starting {app=} @ {host}:{port}")

        # 4. Detect ASGI/FastAPI
        try:
            from fastapi import FastAPI
            is_asgi = isinstance(app, FastAPI)
        except ImportError:
            is_asgi = False

        if is_asgi:
            try:
                import uvicorn
            except ImportError:
                raise RuntimeError("uvicorn is required to serve ASGI apps. Please install uvicorn.")

            uvicorn.run(
                app,
                host=host,
                port=int(port),
                log_level="debug" if debug else "info",
                workers=1,
                reload=debug,
            )
            return

        # 5. Fallback to WSGI servers
        from bottle import run as bottle_run, Bottle
        try:
            from paste import httpserver
        except ImportError:
            httpserver = None

        try:
            from ws4py.server.wsgiutils import WebSocketWSGIApplication
            ws4py_available = True
        except ImportError:
            ws4py_available = False

        if httpserver:
            httpserver.serve(app, host=host, port=int(port))
        elif isinstance(app, Bottle):
            bottle_run(
                app,
                host=host,
                port=int(port),
                debug=debug,
                threaded=threaded,
            )
        else:
            raise TypeError(f"Unsupported WSGI app type: {type(app)}")



    if daemon:
        return asyncio.to_thread(run_server)
    else:
        run_server()