ResponseValue Typing issue

[CoolCat467]

This is not a runtime bug but a typing issue. In flask.typing, ResponseValue does not accept AsyncIterator[str].

I am using quart-trio, which itself uses quart, which uses flask.
quart.templating.stream_template returns AsyncIterator[str], and when running mypy on my project I get the following error:

error: Value of type variable "T_route" of function cannot be "Callable[[], Coroutine[Any, Any, AsyncIterator[str] | Response]]"  [type-var]

Example usage:

__title__ = "Example Server"
__author__ = "CoolCat467"

from quart.templating import stream_template
from quart_trio import QuartTrio
import trio
from os import path, makedirs
import functools
import logging
from quart import Response
from typing import Final
from logging.handlers import TimedRotatingFileHandler
from collections.abc import AsyncIterator
from hypercorn.config import Config
from hypercorn.trio import serve


DOMAIN: str | None = None#getenv("DOMAIN", None)

FORMAT = "[%(asctime)s] [%(levelname)s] %(message)s"

ROOT_FOLDER = trio.Path(path.dirname(__file__))
CURRENT_LOG = ROOT_FOLDER / "logs" / "current.log"

if not path.exists(path.dirname(CURRENT_LOG)):
    makedirs(path.dirname(CURRENT_LOG))

logging.basicConfig(format=FORMAT, level=logging.DEBUG, force=True)
logging.getLogger().addHandler(
    TimedRotatingFileHandler(
        CURRENT_LOG,
        when="D",
        backupCount=60,
        encoding="utf-8",
        utc=True,
        delay=True,
    ),
)


app: Final = QuartTrio(
    __name__,
    static_folder="static",
    template_folder="templates",
)

async def send_error(
    page_title: str,
    error_body: str,
    return_link: str | None = None,
) -> AsyncIterator[str]:
    """Stream error page."""
    return await stream_template(
        "error_page.html.jinja",
        page_title=page_title,
        error_body=error_body,
        return_link=return_link,
    )


async def get_exception_page(code: int, name: str, desc: str) -> Response:
    """Return Response for exception."""
    resp_body = await send_error(
        page_title=f"{code} {name}",
        error_body=desc,
    )
    return Response(resp_body, status=code)


@app.get("/")
async def root_get() -> Response:
    """Main page GET request."""
    return await get_exception_page(404, "Page not found", "Requested content does not exist.")


# Stolen from WOOF (Web Offer One File), Copyright (C) 2004-2009 Simon Budig,
# available at http://www.home.unix-ag.org/simon/woof
# with modifications

# Utility function to guess the IP (as a string) where the server can be
# reached from the outside. Quite nasty problem actually.


def find_ip() -> str:
    """Guess the IP where the server can be found from the network."""
    # we get a UDP-socket for the TEST-networks reserved by IANA.
    # It is highly unlikely, that there is special routing used
    # for these networks, hence the socket later should give us
    # the IP address of the default route.
    # We're doing multiple tests, to guard against the computer being
    # part of a test installation.

    candidates: list[str] = []
    for test_ip in ("192.0.2.0", "198.51.100.0", "203.0.113.0"):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.connect((test_ip, 80))
        ip_addr: str = sock.getsockname()[0]
        sock.close()
        if ip_addr in candidates:
            return ip_addr
        candidates.append(ip_addr)

    return candidates[0]


async def run_async(
    root_dir: str,
    port: int,
    *,
    ip_addr: str | None = None,
    localhost: bool = True,
) -> None:
    """Asynchronous Entry Point."""
    if ip_addr is None:
        ip_addr = "0.0.0.0"  # noqa: S104  # Binding to all interfaces
        if not localhost:
            ip_addr = find_ip()

    try:
        # Add more information about the address
        location = f"{ip_addr}:{port}"

        config = {
            "bind": [location],
            "worker_class": "trio",
        }
        if DOMAIN:
            config["certfile"] = f"/etc/letsencrypt/live/{DOMAIN}/fullchain.pem"
            config["keyfile"] = f"/etc/letsencrypt/live/{DOMAIN}/privkey.pem"
        app.config["SERVER_NAME"] = location

        app.jinja_options = {
            "trim_blocks": True,
            "lstrip_blocks": True,
        }

        app.add_url_rule("/<path:filename>", "static", app.send_static_file)

        config_obj = Config.from_mapping(config)

        proto = "http" if not DOMAIN else "https"
        print(f"Serving on {proto}://{location}\n(CTRL + C to quit)")

        await serve(app, config_obj)
    except OSError:
        logging.error(f"Cannot bind to IP address '{ip_addr}' port {port}")
        sys.exit(1)
    except KeyboardInterrupt:
        logging.info("Shutting down from keyboard interrupt")


def run() -> None:
    """Synchronous Entry Point."""
    root_dir = path.dirname(__file__)
    port = 6002

    hostname: Final = "None"#os.getenv("HOSTNAME", "None")

    ip_address = None
    if hostname != "None":
        ip_address = hostname

    local = True#"--nonlocal" not in sys.argv[1:]

    trio.run(
        functools.partial(
            run_async,
            root_dir,
            port,
            ip_addr=ip_address,
            localhost=local,
        ),
        restrict_keyboard_interrupt_to_checkpoints=True,
    )


def main() -> None:
    """Call run after setup."""
    print(f"{__title__}\nProgrammed by {__author__}.\n")
    try:
        logging.captureWarnings(True)
        run()
    finally:
        logging.shutdown()


if __name__ == "__main__":
    main()

templates/error_page.html.jinja

<!DOCTYPE HTML>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>{{ page_title }}</title>
    <!--<link rel="stylesheet" type="text/css" href="/style.css">-->
  </head>
  <body>
    <div class="content">
      <h1>{{ page_title }}</h1>
      <div class="box">
        <p>
          {{ error_body }}
        </p>
        <br>
        {% if return_link %}
        <a href="{{ return_link }}">Return to previous page</a>
        <br>
        {% endif %}
        <a href="/">Return to main page</a>
      </div>
    </div>
    <footer>
      <i>If you're reading this, the web server was installed correctly.™</i>
      <hr>
      <p>Example Web Server v0.0.0 © CoolCat467</p>
    </footer>
  </body>
</html>

Environment:

• Python version: 3.12
• Flask version: 3.0.0

[davidism]

Flask doesn't accept that though. It would be Quart's response type that should allow that type. If it doesn't, that should be reported to Quart.

[pgjones]

This one is a Flask issue now that Quart is based on Flask. I think the solution is to make the Flask sansio classes Generic over the response type. (Or support async iterators in Flask which could be nice).

@CoolCat467 Something to type: ignore for a while - until I find a nice solution.

[davidism]

Ok, wasn't clear that Quart was passing through Flask's type here.