[REF-723+] Upload with progress and cancellation

[masenf]

Address #1896
REF-723: Progress Display for rx.upload
REF-1037: In prod mode, upload sometimes cannot send delta
REF-611: files: List[rx.UploadFile] does not work with from __future__ import annotations

Improvements

• Pass on_upload_progress kwarg to rx.upload_files and receive a callback as the file is uploading
• rx.upload (and related functions) now accept an id or upload_id kwarg, which allows multiple upload forms to co-exist on the same page and be serviced by different handlers
• rx.cancel_upload(id) allows for in-progress uploads to be cancelled by an event trigger (like on_click or yield from event handler).
• Upload handler returns deltas and events via REST channel, so it works with prod mode and with multiple worker processes. (Uses chunked HTTP so that long running backend processing can still update the frontend)
• Use file.path instead of file.name to support directory uploads
• Support new style list[rx.UploadFile] annotation in upload handler
• Background tasks cannot be upload handler, so explicitly raise an error instead of silently throwing away the deltas.
• Upload passes token and handler via headers instead of munging them into the filename.
• Uploading files never blocks the websocket event channel on the frontend. (Processing the uploaded files on the backed does block the event channel by virtue of locking the state)

Sample Code

from __future__ import annotations
import asyncio
from typing import Dict, List

import reflex as rx
from reflex.components.forms.upload import (
    cancel_upload,
    selected_files,
    clear_selected_files,
)


class UploadState(rx.State):
    _file_data: Dict[str, str] = {}
    upload_progress: Dict[str, int] = {}
    upload_size: Dict[str, int] = {}
    progress_text: Dict[str, str] = {}

    async def handle_upload(self, fx: list[rx.UploadFile]):
        for file in fx:
            upload_data = await file.read()
            self._file_data[file.filename or ""] = upload_data
            yield rx.console_log(f"Uploaded {file.filename}")
            await asyncio.sleep(1)

    def upprog(self, upload_id: str, prog: dict):
        print(upload_id, prog)
        self.upload_progress[upload_id] = prog["loaded"]
        self.upload_size[upload_id] = prog["total"]
        self.progress_text[upload_id] = f"{round(prog.get('progress', 0) * 100)}%"

    def upprog1(self, prog: dict):
        self._upprog("upload1", prog)

    def upprog2(self, prog: dict):
        self._upprog("upload2", prog)

    @rx.var
    def uploaded_files(self) -> List[str]:
        return list(self._file_data)


def upload_form(upload_id):
    return rx.vstack(
        rx.upload(
            rx.vstack(
                rx.button("Select File"),
                rx.text("Drag and drop files here or click to select files"),
            ),
            id=upload_id,
            directory=False,
        ),
        rx.button(
            "Upload",
            on_click=UploadState.handle_upload(  # type: ignore
                rx.upload_files(
                    upload_id=upload_id,
                    on_upload_progress=lambda prog: UploadState.upprog(upload_id, prog),
                )
            ),
        ),
        rx.box(
            rx.foreach(
                selected_files(upload_id),
                rx.text,
            ),
        ),
        rx.text(f"Selected Files: {selected_files(upload_id).length()}"),
        rx.button(
            "Clear",
            on_click=clear_selected_files(upload_id),
        ),
        rx.text(
            "Upload Progress:",
            rx.cond(
                UploadState.progress_text[upload_id],
                UploadState.progress_text[upload_id],
                "0%",
            ),
        ),
        rx.progress(
            value=rx.cond(
                UploadState.upload_progress[upload_id],
                UploadState.upload_progress[upload_id],
                0,
            ),
            min_=0,
            max_=rx.cond(
                UploadState.upload_size[upload_id],
                UploadState.upload_size[upload_id],
                100,
            ),
            width="100%",
        ),
        rx.button("Cancel", on_click=cancel_upload(upload_id)),
        width="50%",
    )


def index():
    return rx.vstack(
        rx.input(value=UploadState.router.session.client_token, is_read_only=True, id="token"),
        rx.hstack(
            upload_form("upload1"),
            upload_form("upload2"),
        ),
        rx.vstack(
            rx.foreach(
                UploadState.uploaded_files,
                lambda f: rx.text(f),
            ),
        ),
    )


app = rx.App(state=UploadState)
app.add_page(index)
app.compile()

☝️ Separate, independently operating upload forms!!

[Lendemor]

What's the reason for excluding background tasks as upload handler?
Does it cause issues somewhere?

[masenf]

The problem is that a background task relies on websocket to send the deltas to the frontend, but the upload event comes via REST and might hit an app instance that is not managing the websocket associated with the session that made the upload. So if the upload handler is a background task, and the response is handled by a different instance the deltas wont be sent and the user will file a bug or just be confused why things work in dev mode and not work with redis.

Ultimately want to solve this with REF-1073, but it's non-trivial.

[picklelo]

Nice!

[picklelo]

Just my preference - but can we remove the for and have this just be upload_file or get_upload_file. I think it matches the rest of our API better as we don't do this in other places

[picklelo]

Similarly can this become selected_files or get_selected_files

[picklelo]

It will match this API also, since this doesn't have the for