feat: add chunked file upload support Streaming Upload API (rx.upload_files_chunk)

Implement chunked/streaming file uploads to handle large files without
loading them entirely into memory. Moves upload handling logic from
app.py to event.py, adds chunked upload JS helpers, and updates the
upload component to support the new upload_files_chunk API. Includes
unit and integration tests for chunked upload, cancel, and streaming.

Author: FarhanAliRaza

pyi_hashes.json

@@ -1,5 +1,5 @@
 {
-  "reflex/__init__.pyi": "0a3ae880e256b9fd3b960e12a2cb51a7",
+  "reflex/__init__.pyi": "70485139882c5c114c121445a24c7b28",
   "reflex/components/__init__.pyi": "ac05995852baa81062ba3d18fbc489fb",
   "reflex/components/base/__init__.pyi": "16e47bf19e0d62835a605baa3d039c5a",
   "reflex/components/base/app_wrap.pyi": "22e94feaa9fe675bcae51c412f5b67f1",
@@ -19,7 +19,7 @@
   "reflex/components/core/helmet.pyi": "43f8497c8fafe51e29dca1dd535d143a",
   "reflex/components/core/html.pyi": "86eb9d4c1bb4807547b2950d9a32e9fd",
   "reflex/components/core/sticky.pyi": "cb763b986a9b0654d1a3f33440dfcf60",
-  "reflex/components/core/upload.pyi": "6dc28804a6dddf903e31162e87c1b023",
+  "reflex/components/core/upload.pyi": "58c023b9149635894331528bf29eaf13",
   "reflex/components/core/window_events.pyi": "af33ccec866b9540ee7fbec6dbfbd151",
   "reflex/components/datadisplay/__init__.pyi": "52755871369acbfd3a96b46b9a11d32e",
   "reflex/components/datadisplay/code.pyi": "b86769987ef4d1cbdddb461be88539fd",

reflex/.templates/web/utils/helpers/upload.js

@@ -1,47 +1,11 @@
 import JSON5 from "json5";
 import env from "$/env.json";
 
-/**
- * Upload files to the server.
- *
- * @param state The state to apply the delta to.
- * @param handler The handler to use.
- * @param upload_id The upload id to use.
- * @param on_upload_progress The function to call on upload progress.
- * @param socket the websocket connection
- * @param extra_headers Extra headers to send with the request.
- * @param refs The refs object to store the abort controller in.
- * @param getBackendURL Function to get the backend URL.
- * @param getToken Function to get the Reflex token.
- *
- * @returns The response from posting to the UPLOADURL endpoint.
- */
-export const uploadFiles = async (
-  handler,
-  files,
-  upload_id,
-  on_upload_progress,
-  extra_headers,
-  socket,
-  refs,
-  getBackendURL,
-  getToken,
-) => {
-  // return if there's no file to upload
-  if (files === undefined || files.length === 0) {
-    return false;
-  }
-
-  const upload_ref_name = `__upload_controllers_${upload_id}`;
-
-  if (refs[upload_ref_name]) {
-    console.log("Upload already in progress for ", upload_id);
-    return false;
-  }
-
+const trackUploadResponse = (socket) => {
   // Track how many partial updates have been processed for this upload.
   let resp_idx = 0;
-  const eventHandler = (progressEvent) => {
+
+  return (progressEvent) => {
     const event_callbacks = socket._callbacks.$event;
     // Whenever called, responseText will contain the entire response so far.
     const chunks = progressEvent.event.target.responseText.trim().split("\n");
@@ -73,22 +37,32 @@ export const uploadFiles = async (
       }
     });
   };
+};
 
-  const controller = new AbortController();
-  const formdata = new FormData();
+const sendUploadRequest = async ({
+  handler,
+  upload_id,
+  on_upload_progress,
+  extra_headers,
+  refs,
+  getToken,
+  formdata,
+  url,
+  responseHandler,
+}) => {
+  const upload_ref_name = `__upload_controllers_${upload_id}`;
 
-  // Add the token and handler to the file name.
-  files.forEach((file) => {
-    formdata.append("files", file, file.path || file.name);
-  });
+  if (refs[upload_ref_name]) {
+    return false;
+  }
+
+  const controller = new AbortController();
 
-  // Send the file to the server.
   refs[upload_ref_name] = controller;
 
   return new Promise((resolve, reject) => {
     const xhr = new XMLHttpRequest();
 
-    // Set up event handlers
     xhr.onload = function () {
       if (xhr.status >= 200 && xhr.status < 300) {
         resolve({
@@ -112,42 +86,36 @@ export const uploadFiles = async (
       reject(new Error("Upload aborted"));
     };
 
-    // Handle upload progress
     if (on_upload_progress) {
       xhr.upload.onprogress = function (event) {
         if (event.lengthComputable) {
-          const progressEvent = {
+          on_upload_progress({
             loaded: event.loaded,
             total: event.total,
             progress: event.loaded / event.total,
-          };
-          on_upload_progress(progressEvent);
+          });
         }
       };
     }
 
-    // Handle download progress with streaming response parsing
-    xhr.onprogress = function (event) {
-      if (eventHandler) {
-        const progressEvent = {
+    if (responseHandler) {
+      xhr.onprogress = function (event) {
+        responseHandler({
           event: {
             target: {
               responseText: xhr.responseText,
             },
           },
           progress: event.lengthComputable ? event.loaded / event.total : 0,
-        };
-        eventHandler(progressEvent);
-      }
-    };
+        });
+      };
+    }
 
-    // Handle abort controller
     controller.signal.addEventListener("abort", () => {
       xhr.abort();
     });
 
-    // Configure and send request
-    xhr.open("POST", getBackendURL(env.UPLOAD));
+    xhr.open("POST", url);
     xhr.setRequestHeader("Reflex-Client-Token", getToken());
     xhr.setRequestHeader("Reflex-Event-Handler", handler);
     for (const [key, value] of Object.entries(extra_headers || {})) {
@@ -168,3 +136,99 @@ export const uploadFiles = async (
       delete refs[upload_ref_name];
     });
 };
+
+/**
+ * Upload files to the server.
+ *
+ * @param handler The handler to use.
+ * @param upload_id The upload id to use.
+ * @param on_upload_progress The function to call on upload progress.
+ * @param extra_headers Extra headers to send with the request.
+ * @param socket The websocket connection.
+ * @param refs The refs object to store the abort controller in.
+ * @param getBackendURL Function to get the backend URL.
+ * @param getToken Function to get the Reflex token.
+ *
+ * @returns The response from posting to the upload endpoint.
+ */
+export const uploadFiles = async (
+  handler,
+  files,
+  upload_id,
+  on_upload_progress,
+  extra_headers,
+  socket,
+  refs,
+  getBackendURL,
+  getToken,
+) => {
+  if (files === undefined || files.length === 0) {
+    return false;
+  }
+
+  const formdata = new FormData();
+
+  files.forEach((file) => {
+    formdata.append("files", file, file.path || file.name);
+  });
+
+  return sendUploadRequest({
+    handler,
+    upload_id,
+    on_upload_progress,
+    extra_headers,
+    refs,
+    getToken,
+    formdata,
+    url: getBackendURL(env.UPLOAD),
+    responseHandler: trackUploadResponse(socket),
+  });
+};
+
+/**
+ * Upload files to the streaming chunk endpoint.
+ *
+ * @param handler The handler to use.
+ * @param files The files to upload.
+ * @param upload_id The upload id to use.
+ * @param on_upload_progress The function to call on upload progress.
+ * @param extra_headers Extra headers to send with the request.
+ * @param _socket The websocket connection.
+ * @param refs The refs object to store the abort controller in.
+ * @param getBackendURL Function to get the backend URL.
+ * @param getToken Function to get the Reflex token.
+ *
+ * @returns The response from posting to the chunk upload endpoint.
+ */
+export const uploadFilesChunk = async (
+  handler,
+  files,
+  upload_id,
+  on_upload_progress,
+  extra_headers,
+  _socket,
+  refs,
+  getBackendURL,
+  getToken,
+) => {
+  if (files === undefined || files.length === 0) {
+    return false;
+  }
+
+  const formdata = new FormData();
+
+  files.forEach((file) => {
+    formdata.append("files", file, file.path || file.name);
+  });
+
+  return sendUploadRequest({
+    handler,
+    upload_id,
+    on_upload_progress,
+    extra_headers,
+    refs,
+    getToken,
+    formdata,
+    url: getBackendURL(env.UPLOAD_CHUNK),
+  });
+};

reflex/.templates/web/utils/state.js

@@ -20,7 +20,7 @@ import {
 } from "$/utils/context";
 import debounce from "$/utils/helpers/debounce";
 import throttle from "$/utils/helpers/throttle";
-import { uploadFiles } from "$/utils/helpers/upload";
+import { uploadFiles, uploadFilesChunk } from "$/utils/helpers/upload";
 
 // Endpoint URLs.
 const EVENTURL = env.EVENT;
@@ -418,21 +418,27 @@ export const applyEvent = async (event, socket, navigate, params) => {
  */
 export const applyRestEvent = async (event, socket, navigate, params) => {
   let eventSent = false;
-  if (event.handler === "uploadFiles") {
-    if (event.payload.files === undefined || event.payload.files.length === 0) {
+  if (event.handler === "uploadFiles" || event.handler === "uploadFilesChunk") {
+    const filePayloadKey = event.payload.upload_param_name || "files";
+    const uploadFilesPayload =
+      event.payload.files ?? event.payload[filePayloadKey];
+
+    if (uploadFilesPayload === undefined || uploadFilesPayload.length === 0) {
       // Submit the event over the websocket to trigger the event handler.
       return await applyEvent(
-        ReflexEvent(event.name, { files: [] }),
+        ReflexEvent(event.name, { [filePayloadKey]: [] }),
         socket,
         navigate,
         params,
       );
     }
 
     // Start upload, but do not wait for it, which would block other events.
-    uploadFiles(
+    const uploadFn =
+      event.handler === "uploadFilesChunk" ? uploadFilesChunk : uploadFiles;
+    uploadFn(
       event.name,
-      event.payload.files,
+      uploadFilesPayload,
       event.payload.upload_id,
       event.payload.on_upload_progress,
       event.payload.extra_headers,

reflex/__init__.py

@@ -301,6 +301,8 @@
         "event",
         "EventChain",
         "EventHandler",
+        "UploadChunk",
+        "UploadChunkIterator",
         "call_script",
         "call_function",
         "run_script",
@@ -320,6 +322,7 @@
         "set_value",
         "stop_propagation",
         "upload_files",
+        "upload_files_chunk",
         "window_alert",
     ],
     "istate.storage": [
@@ -348,6 +351,7 @@
 _SUBMODULES: set[str] = {
     "components",
     "app",
+    "uploads",
     "style",
     "admin",
     "base",