ui.upload frontend component

[PeterMarkuson]

Hi guys,

recently I am trying to develop own components for my project. I wanted to have a simpler and smaller ui.upload panel. Basically it should just be a button with a counter. It should allow to upload several files by clicking it more than once and it should allow to save data into memory for further processing before saving to disk. Also, I just wanted to have ONE uploader in my project, so with this component I can still choose to show the original ui.upload or my button solution.

This is my code so far:

from datetime import datetime
from pathlib import Path
from nicegui import ui, events
import asyncio
import time

#TODO Replace print statements with logger

class Uploader(ui.element):
    '''
    Uploader component utilizing ui.upload in background, but providing a standard button for operation.
    The component is callable and (sync/async) iterable.
    In case of longer uploads the button is replaced with a spinner to show that work is in progress.

    root_path = where to save the upload
    file_name = pre-defined name or auto select if None
    use_type_as_suffix = the mime type will become the universal suffix of the file (e.g. .image or .video instead of .jpg or .mp4)
    auto_save = save automatically on disk if True or keep in memory for further processing if False
    accept = what filetypes to accept
    on_success = additional callback called after successful upload
    on_recect = additional callback called if rejected
    min_size = show the component as a standard button if True otherwise use the original ui.uploader design
    label = Label of the button
    icon = icon of the button
    show_badge = Show a counter for files uploaded if True
    positive_msg = Standard message after upload
    negative_msg = Standard message after rejection
    '''

    def __init__(self, *, root_path, file_name=None, use_type_as_suffix=False, auto_save=False, accept="image/*, video/*, audio/*",
                 on_success=None, on_reject=None, min_size=True, label=None, icon="cloud_upload", show_badge=True,
                 positive_msg="Upload erfolgreich", negative_msg="Upload abgelehnt"):
        super().__init__()

        self.root_path = root_path
        if not Path(self.root_path).exists():
            raise FileNotFoundError(f"Root path {self.root_path} not existing")

        # save all parameters and state
        self.file_name = file_name
        self.auto_save = auto_save
        self.accept = accept
        self.saved_to_disk = False
        self.saved_to_memory = False
        self.use_type_as_suffix = use_type_as_suffix
        self.on_success = on_success
        self.on_reject = on_reject
        self.positive_msg = positive_msg
        self.negative_msg = negative_msg
        self.min_size=min_size
        self.label = label
        self.icon = icon
        self.show_badge=show_badge

        with ui.button(self.label, icon=self.icon, on_click=self.pick_file) as self.upload_button:
            self.button_badge = ui.badge("0", color="accent").props("floating")
            self.button_badge.set_visibility(self.show_badge)

        self.upload_spinner = ui.spinner("facebook",size="3em",color="info")
        self.upload_spinner.set_visibility(False)

        self.upload_manager = ui.upload(label=self.label, multiple=False, auto_upload=True,
                                        on_upload=self.handle_upload, on_rejected=self.handle_reject)
        self.upload_manager.ignores_events_when_hidden = False
        self.upload_manager.props(f"no-thumbnails accept='{self.accept}'")
        self.upload_manager.on("uploading",self.switch_upload_status)
        self.upload_manager.on("uploaded",self.switch_upload_status)

        if self.min_size:
            self.upload_manager.set_visibility(False)
        else:
            self.upload_button.set_visibility(False)

    async def pick_file(self):
        await self.upload_manager.run_method("pickFiles")

    async def switch_upload_status(self):
        if not self.upload_manager.visible:
            self.upload_button.set_visibility(not self.upload_button.visible)
            self.upload_spinner.set_visibility(not self.upload_spinner.visible)

    async def handle_reject(self, e: events.UploadEventArguments):
        ui.notify(self.negative_msg, color="negative")

        if not self.on_reject:
            return
        elif asyncio.iscoroutinefunction(self.on_reject):
            await self.on_reject(self)
        else:
            self.on_reject(self)

    async def handle_upload(self, e: events.UploadEventArguments):
        self.file_type = e.type
        self.saved_file_name = self.file_name or f"{datetime.now()}".replace(" ", "_")
        self.full_name = f"{self.root_path}/{self.saved_file_name}.{e.type.split('/')[0 if self.use_type_as_suffix else -1]}"
        self.file_content = e.content
        self.saved_to_memory = True
        if self.auto_save:
            await self.save_to_disk(self.full_name, self.file_content)
        else:
            ui.notify(self.positive_msg, color="positive")
            print("Upload saved to memory")

        if not self.on_success:
            return
        if asyncio.iscoroutinefunction(self.on_success):
            await asyncio.sleep(0.5)
            await self.on_success()
        else:
            time.sleep(0.5)
            self.on_success()

    async def save_to_disk(self, full_name, content):
        with open(Path(full_name), "wb") as file:
            file.write(content.read())
            self.saved_to_disk = True
            print(f"Upload saved to {full_name}")
            ui.notify(self.positive_msg, color="positive")

    def __call__(self) -> dict:
        return {
            "file_type": self.file_type,
            "file_content": self.file_content,
            "saved_file_name": self.saved_file_name,
            "full_name": self.full_name
        }

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, exc_tb):
        self = None

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        self = None

And it is used as easy as

with Uploader(root_path=self.attachment_path, icon="attachment", on_success=self.after_upload_attachment) as self.attachment_uploader:
    self.attachment_uploader.button_badge.set_text(len(self.attachments))

where self.after_upload_attachment is my callback to process the upload. You can easily modify all used niceGUI components like the badge, button or the original upload panel (like in this example I set the badge text).

It shows up like this (screenshot from my app):

On mobile devices (tested on Android) it will allow the user to pick either the camera app or voice recorder app or file picker to provide the data. On Windows it will start the file picker.

BR/Peter

[PeterMarkuson]

After further testing this, I would like to ask for a best practice advice on how to deal with session cuts in case the selection of the file is taking too long and the session is cut.

With my component: I click on the icon and the file browser opens. If it takes too long I see in the logs
client id 3fbf63b1-ea00-4d90-85b4-dd13749f770e disconnect

and after I return from the filebrowser this:
http://IP:PORT/_nicegui/client/3fbf63b1-ea00-4d90-85b4-dd13749f770e/upload/35 not found

the next client connect is from a different id and the old id is never connecting again. The page knows the state of the fields, so that is not an issue. Its specifically about how to handle uploads.

Any advice would be highly appreciated

BR/Peter

[rodja]

This sounds like #2286. The feature request #2410 has quite some upvotes but unfortunately there is no implementation yet.

[PeterMarkuson]

Thank you for your swift answer. Really appreciate your support. I am already aware of these topics and just hoping there is a common workaround.

I could imagine some experienced people ran into the same and solved it 🙂

Like how to detect that the connection was lost and try to link the loose pieces together again somehow, so that the upload will execute.

BR/Peter