Function UploadFile

tests/integration/test_upload.py:24–339

App for testing dynamic routes.

()

def UploadFile():
    """App for testing dynamic routes."""
    import shutil

    import reflex as rx

    LARGE_DATA = "DUMMY" * 1024 * 512

    class UploadState(rx.State):
        upload_done: rx.Field[bool] = rx.field(False)
        event_order: rx.Field[list[str]] = rx.field([])
        progress_dicts: rx.Field[list[dict]] = rx.field([])
        stream_progress_dicts: rx.Field[list[dict]] = rx.field([])
        disabled: rx.Field[bool] = rx.field(False)
        large_data: rx.Field[str] = rx.field("")
        quaternary_names: rx.Field[list[str]] = rx.field([])
        quaternary_field: rx.Field[str] = rx.field("")
        stream_chunk_records: rx.Field[list[str]] = rx.field([])
        stream_completed_files: rx.Field[list[str]] = rx.field([])
        stream_field: rx.Field[str] = rx.field("")

        @rx.event
        async def handle_upload(self, files: list[rx.UploadFile]):
            self.upload_done = False
            for file in files:
                upload_data = await file.read()
                if not file.name:
                    continue
                local_file = rx.get_upload_dir() / file.name
                local_file.parent.mkdir(parents=True, exist_ok=True)
                local_file.write_bytes(upload_data)
            self.upload_done = True

        @rx.event
        async def handle_upload_secondary(self, files: list[rx.UploadFile]):
            self.upload_done = False
            for file in files:
                upload_data = await file.read()
                if not file.name:
                    continue
                local_file = rx.get_upload_dir() / file.name
                local_file.parent.mkdir(parents=True, exist_ok=True)
                local_file.write_bytes(upload_data)
                self.large_data = LARGE_DATA
                yield UploadState.chain_event

        @rx.event
        def upload_progress(self, progress):
            assert progress
            print(self.event_order)
            self.progress_dicts.append(progress)

        @rx.event
        def chain_event(self):
            assert self.large_data == LARGE_DATA
            self.large_data = ""
            self.upload_done = True
            self.event_order.append("chain_event")