How to call a function after file upload?

[IEC1]

Hi there,

I've built a ComponentState (see below) that returns an rx.upload.
After files have uploaded, I want to call an rx.event (also passing the files to this event)

I've tried:

• Having the on_upload_complete function as a normal function and an rx.event
• Saving the on_upload_complete function to a state var and calling it in the handle_file_upload function
• Using on_change for the rx.upload

Is this the right way to approach this?

from typing import Callable, Optional
import reflex as rx


class FileInput(rx.ComponentState):
    files: rx.Field[list[rx.UploadFile]] = rx.field(default_factory=list)

    @rx.event
    def set_on_upload_callback(self, callback: Callable):
        self._on_upload_complete = callback

    @rx.event
    async def handle_file_upload(
        self,
        files: list[rx.UploadFile],
        on_upload_complete: Optional[list[Callable]]
    ):
        if not files:
            return

        self.files = []

        for file in files:
            data = await file.read()
            path = rx.get_upload_dir() / file.name
            with path.open("wb") as f:
                f.write(data)
            self.files.append(file)

        if on_upload_complete:
            for callback in on_upload_complete:
                if isinstance(callback, Callable):
                    await callback(files)

    @rx.var
    def file_names(self) -> list[str]:
        return [file.name for file in self.files]

    @rx.var
    def num_files(self) -> int:
        return len(self.files)

    @rx.var
    def selection_message(self) -> str:
        if self.num_files == 0:
            return "No File Selected"
        elif self.num_files == 1:
            return self.files[0].name
        else:
            return f"{self.num_files} Files Uploaded"

    @classmethod
    def get_component(cls, id: str, **props):
        on_drop = [
            cls.handle_file_upload(
                rx.upload_files(f"{id}-file-input")
            )
        ]

        user_on_drop = props.pop("on_upload_complete", None)
        if user_on_drop is not None:
            on_drop.append(user_on_drop)

        return rx.box(
            rx.upload(
                rx.hstack(
                    rx.button(
                        "Choose Files",
                    ),
                    rx.box(
                        cls.selection_message,
                        max_width="20ch",
                        width="300px",
                        height="2.5rem",
                        padding="0rem 0.5rem",
                        align_content="center",
                        background_color="white",
                        color="var(--gray-11)",
                        white_space="nowrap",
                        overflow="hidden",
                        text_overflow="ellipsis",
                    ),
                ),
                id=f"{id}-file-input",
                accept=props.get("file_types", {}),
                on_drop=on_drop,
            )
        )

Thanks

[masenf]

Instead of trying to directly called the chained events, just return/yield the events directly from the backend and they will be queued on the frontend and sent after the upload handler completes.

So change

        if on_upload_complete:
            for callback in on_upload_complete:
                if isinstance(callback, Callable):
                    await callback(files)

to

        if on_upload_complete:
            return on_upload_complete  # or return self._on_upload_complete

Event handlers can return a single EventHandler or EventSpec or a sequence of EventHandler or EventSpec to chain them.

[IEC1]

Hi @masenf,

Thanks for that solution.

The handle_file_upload didn't allow other arguments such as on_upload_complete as per the following error:

reflex.utils.exceptions.EventHandlerArgTypeMismatchError: Event handler expects list[reflex.app.UploadFile] for argument on_upload_complete but got typing.Optional[list[typing.Callable]] as annotated in FileInput_n1.handle_file_upload instead.

Because of this, I'm now trying to save the rx.event in the ComponentState to return later.
I've tried the following from get_component but it doesn't save:

setattr(cls, "_on_upload_complete", on_upload_complete)

cls.__fields__["_on_upload_complete"].default = on_upload_complete

cls._on_upload_complete = on_upload_complete

cls.set_on_upload_callback(on_upload_complete)
"""
@rx.event
def set_on_upload_callback(self, callback: Callable):
    self._on_upload_complete = callback
"""

[masenf]

if you need the callback for get_component, can you just save it as a ClassVar instead of trying to save it on the instance as a field default?

[MatthewCollinsNZ]

Hi @masenf,
We are trying to create a generic upload component state which can be reused throughout all our apps, the idea here is to have the component be able to take a file on drop, and after the upload is complete push the files through to another event which was passed in.

We have tried two ways to do this:

1. Appending the external event to the on drop so it runs it after, (this does either calls the external function at the wrong time or does not at all)
2. Passing the un_upload_complete event into the handle file upload event (fails because it seems that its expecting rx.upload file though its not quite clear in the error where its failing)

Below I have created two generic component states to test this, please take a look

import reflex as rx
from typing import Callable, Optional


class FileInput1(rx.ComponentState):
    files: list[rx.UploadFile] = []
    current_state = "Empty"

    @rx.event
    async def handle_file_upload(self, files: list[rx.UploadFile]):
        """Handle file upload and store the files in state."""
        print("Handling file upload...")
        print(f'Files received: {files}')

        self.current_state = "Uploading"

        if not files:
            return

        # Save files to disk
        for file in files:
            path = rx.get_upload_dir() / file.name
            with path.open("wb") as f:
                f.write(file.file.getbuffer())

        self.files = files
        print("Files uploaded:", [file.name for file in self.files])
        self.current_state = "Uploaded"

    @classmethod
    def get_component(cls, on_upload_complete: Optional[Callable] = None, **props):
        """Create FileInput with optional on_upload_complete callback."""
        
        # Build the event chain
        on_drop = [
            cls.handle_file_upload(rx.upload_files(f"{props.get('id', 'file-input')}-file-input")),
        ]
        
        # Append the on_upload_complete callback if provided
        if on_upload_complete:
            print("Adding on_upload_complete callback")
            on_drop.append(on_upload_complete(cls.files))

        return rx.upload(
            rx.button("Choose Files"),
            id=f"{props.get('id', 'file-input')}-file-input",
            on_drop=on_drop,
        )


class FileInput2(rx.ComponentState):
    files: list[rx.UploadFile] = []
    current_state = "Empty"

    @rx.event
    async def handle_file_upload(self, files: list[rx.UploadFile], on_upload_complete: Optional[Callable] = None):
        """Handle file upload and store the files in state."""
        print("Handling file upload...")
        print(f'Files received: {files}')

        self.current_state = "Uploading"

        if not files:
            return

        # Save files to disk
        for file in files:
            path = rx.get_upload_dir() / file.name
            with path.open("wb") as f:
                f.write(file.file.getbuffer())

        self.files = files
        print("Files uploaded:", [file.name for file in self.files])
        self.current_state = "Uploaded"

        # Call the on_upload_complete callback if provided
        if on_upload_complete:
            print("Calling on_upload_complete callback")
            await on_upload_complete(self.files)

    @classmethod
    def get_component(cls, on_upload_complete: Optional[Callable] = None, **props):
        """Create FileInput with optional on_upload_complete callback."""
        
        # Build the event chain
        on_drop = [
            cls.handle_file_upload(rx.upload_files(f"{props.get('id', 'file-input')}-file-input")),
        ]
        
        # Append the on_upload_complete callback if provided
        if on_upload_complete:
            print("Adding on_upload_complete callback")
            on_drop.append(on_upload_complete(cls.files))

        return rx.upload(
            rx.button("Choose Files"),
            id=f"{props.get('id', 'file-input')}-file-input",
            on_drop=on_drop,
        )