Interactive web app without javascript; using css and long polling to dynamically update the page. async, ASGI. Built using Starlette framework and Uvicorn server

requirements.txt

# back when i wrote this i didnt add any version info
# so i wasnt able to run it almost a year later

# anyway heres last working versions as of 04/14/2020

# python 3.8.2

starlette==0.11.4 # old, Mar 18, 2019
uvicorn==0.11.3 # this is the newest at time of writing

# everything else is a pip freeze

click==7.1.1
h11==0.9.0
httptools==0.1.
uvloop==0.14.0
websockets==8.1

stream_server.py

#!/usr/bin/env python

from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.requests import HTTPConnection
import asyncio
import uuid
import enum
import random
import string

app = Starlette()
app.debug = True

muh_globals = {}


async def never_return():
    while True:
        await asyncio.sleep(10 ** 10)


async def instant_return():
    pass


async def cancel_task(task):
    task.cancel()
    # await asyncio.wait_for(task, 15)
    asyncio.create_task(asyncio.wait_for(task, 15))


class AsgiState(enum.Enum):
    NOTREADY = enum.auto()
    OPEN = enum.auto()
    CLOSED = enum.auto()


# handle is this handler
# handler is subclass handler
# handle is called first, which then calls handler


@app.route("/asgi")
class ASGIHTTPApp:
    # Implement ASGI spec

    __send_errors = {
        AsgiState.NOTREADY: asyncio.InvalidStateError,
        AsgiState.OPEN: lambda: False,
        AsgiState.CLOSED: asyncio.CancelledError,
    }

    __receive_errors = {
        AsgiState.NOTREADY: lambda: False,
        AsgiState.OPEN: asyncio.InvalidStateError,
        AsgiState.CLOSED: asyncio.CancelledError,
    }

    def __init__(self, scope, handler=instant_return):
        self.scope = scope
        self.__handler = handler

        self.__state = AsgiState.NOTREADY
        self.__lock = asyncio.Lock()

    async def __call__(self, receive, sender):
        self.__receive = receive
        self.__sender = sender

        try:
            await self.__handle()
        except asyncio.CancelledError:
            pass

        print("actually ended thank goodness")

    async def sender(self, data):
        async with self.__lock:
            error = ASGIHTTPApp.__send_errors[self.__state]()
            if error:
                raise error

            print("\u2588", end="")
            await self.__sender(data)

    async def __handle(self):
        body = b""

        async def handle_start(data):
            error = ASGIHTTPApp.__receive_errors[self.__state]()
            if error:
                raise error

            nonlocal body
            body += data.get("body", b"")
            if not data.get("more_body", False):
                self.__state = AsgiState.OPEN
                self.__task = asyncio.create_task(self.__handler(body))

        async def handle_end(data):
            self.__state = AsgiState.CLOSED
            try:
                await cancel_task(self.__task)
            except (NameError, AttributeError):
                print("Http disconnect before request finalized")

            raise asyncio.CancelledError()

        # the handlers must exit timely
        event_lookup = {"http.request": handle_start, "http.disconnect": handle_end}

        while True:
            res = await self.__receive()
            # raise error if event not able to be handled, as part of spec
            await event_lookup[res["type"]](res)


class StreamingASGIApp(ASGIHTTPApp):
    KEEP_ALIVE_PERIOD = 5

    def __init__(self, scope, handler=instant_return):
        super().__init__(scope, self.__handle)
        self.__handler = handler

    async def __handle(self, body=b""):
        try:
            await self.start_response()
            self.__keep_alive_task = asyncio.create_task(self.__keep_alive_loop())

            self.__task = asyncio.create_task(self.__handler(body))
            await asyncio.wait_for(self.__task, timeout=None)
        except asyncio.CancelledError:
            pass
        except Exception as e:
            print(repr(e))
            await self.send(f"\n\nException: {repr(e)}")
            raise
        finally:
            try:
                await cancel_task(self.__keep_alive_task)
            except (NameError, AttributeError):
                pass

            try:
                await cancel_task(self.__task)
            except (NameError, AttributeError):
                pass

            await self.end_response()

    # keepalive

    async def __keep_alive_loop(self):
        while True:
            await self.send(None, " ")
            await asyncio.sleep(StreamingASGIApp.KEEP_ALIVE_PERIOD)

    # wrappers for sending

    async def start_response(self, status=200, headers=None):
        if headers is None:
            headers = []

        headers.append([b"Transfer-Encoding", b"chunked"])
        headers.append([b"Content-Type", b"text/html;charset=utf-8"])

        data = {"type": "http.response.start", "status": status, "headers": headers}

        await self.sender(data)

    async def send(self, body=None, end="\n"):
        if body is None:
            body = ""

        print(body, end=end)

        body = body + end
        body = bytes(body, "utf-8")

        data = {"type": "http.response.body", "body": body, "more_body": True}

        await self.sender(data)

    async def end_response(self):
        data = {"type": "http.response.body"}  # more_body is false by default

        await self.sender(data)


@app.route("/")
class ActualApp(StreamingASGIApp):
    def __init__(self, scope):
        super().__init__(scope, self.__handle)

    async def __handle(self, data=b""):
        self.next_uuid()

        try:
            await self.send_preamble()
            await self.start_block()
            await self.send(f"Your UUID is: {self.uuid}")
            await self.send_block()

            self.__task = asyncio.create_task(never_return())
            await asyncio.wait_for(self.__task, None)
        finally:
            await self.send("closing...")

            await self.end_block()
            await self.send_postamble()
            self.del_uuid()

    # uuid

    def del_uuid(self):
        try:
            old_uuid = self.uuid
        except (NameError, AttributeError):
            pass
        else:
            del muh_globals[old_uuid]

    def next_uuid(self):
        self.del_uuid()

        self.uuid = "".join([random.choice(string.ascii_lowercase) for x in range(10)])
        muh_globals[self.uuid] = self

        return self.uuid

    # block

    async def next_block(self):
        await self.end_block()
        await self.send(f"<style>.{self.uuid} " + r"{display: none;}</style>")
        self.next_uuid()
        await self.start_block()
        await self.send(f"Your UUID is: {self.uuid}")

    async def start_block(self):
        await self.send(f'<div class="{self.uuid}">')

    async def end_block(self):
        await self.send("</div>")
        # await self.send(f"Just ended the {self.uuid} block")

    # start + end response

    async def send_preamble(self):
        await self.send(
            f"<html><head><title>{self.uuid}</title>"
            # + r"<style>body{ white-space: pre-line; }</style>"
            + "</head><body>"
        )

    async def send_postamble(self):
        await self.send(r"</body></html>")

    # actual app

    async def send_buttons(self, names):
        styles = []
        bodies = []

        for name in names:
            styles.append(
                f'.{self.uuid} .{name}:active {{ background-image: url("/api/{self.uuid}/{name}"); }}'
            )
            bodies.append(f'<input type="button" class="{name}" value="{name}">')

        style = "\n".join(styles)
        style = f"<style>{style}</style>"

        body = "\n".join(bodies)
        body = f"<div>{body}</div>"

        await self.send(style)
        await self.send(body)

    async def send_block(self):
        buttons = ["close", "asdf", "qwerty", "foobar"]
        await self.send_buttons(buttons)

    # events

    async def end(self):
        await self.end_block()
        await cancel_task(self.__task)  # enter finally in handle

    async def handle_event(self, event):
        if event.strip().casefold() == "close".casefold():
            return await self.end()

        await self.next_block()
        await self.send("<br>")
        await self.send_block()
        await self.send(f"<br>Last event recieved: {event}")


@app.route("/api/{uuid}/{event}")
class StreamingASGIApp(StreamingASGIApp):
    def __init__(self, scope):
        super().__init__(scope, self.__handler)

    async def __handler(self, body=b""):
        request = HTTPConnection(self.scope)

        uuid = request.path_params["uuid"]
        event = request.path_params["event"]

        try:
            await self.send(
                f"<html><head><title>Event Handler</title>"
                + r"<style>body{ white-space: pre-line; }</style></head><body>"
            )

            await self.send(f"Loading event {event} on uuid {uuid}")

            streamer = muh_globals.get(uuid)
            if streamer is None:
                await self.send("Uuid not found")
                raise HTTPException(404)

            await self.send(f"Uuid found: {str(streamer)}")
            await streamer.handle_event(event)
            await self.send(f"Handled successfully")

        finally:
            await self.send(r"</body></html>")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)