feat(files): stream uploads to disk instead of buffering

marimo/_server/api/endpoints/file_explorer.py

@@ -4,11 +4,15 @@
 from typing import TYPE_CHECKING
 
 from starlette.authentication import requires
+from starlette.exceptions import HTTPException
 
 from marimo import _loggers
 from marimo._server.api.deps import AppState
 from marimo._server.api.utils import parse_multipart_request, parse_request
-from marimo._server.files.os_file_system import OSFileSystem
+from marimo._server.files.os_file_system import (
+    OSFileSystem,
+    UploadTooLargeError,
+)
 from marimo._server.models.files import (
     FileCopyRequest,
     FileCopyResponse,
@@ -120,16 +124,24 @@ async def create_file_or_directory(
                         $ref: "#/components/schemas/FileCreateResponse"
     """
     try:
-        parsed = await parse_multipart_request(
+        async with parse_multipart_request(
             request, FileCreateMultipartRequest
-        )
-        info = file_system.create_file_or_directory(
-            parsed.body.path,
-            parsed.body.type,
-            parsed.body.name,
-            parsed.files.get("file"),
-        )
+        ) as parsed:
+            upload = parsed.files.get("file")
+            # Directories and the default-template notebook take the
+            # in-memory path; only real file content streams.
+            if upload is not None and parsed.body.type in ("file", "notebook"):
+                info = await file_system.stream_create_file(
+                    parsed.body.path, parsed.body.name, upload
+                )
+            else:
+                info = file_system.create_file_or_directory(
+                    parsed.body.path, parsed.body.type, parsed.body.name, None
+                )
         return FileCreateResponse(success=True, info=info)
+    except UploadTooLargeError as e:
+        LOGGER.warning(f"Rejected oversize upload: {e}")
+        raise HTTPException(status_code=413, detail=str(e)) from e
     except Exception as e:
         LOGGER.error(f"Error creating file or directory: {e}")
         return FileCreateResponse(success=False, message=str(e))

marimo/_server/api/utils.py

@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import webbrowser
+from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from pathlib import Path
 from shutil import which
@@ -25,6 +26,9 @@
 from marimo._utils.parse_dataclass import parse_raw
 
 if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from starlette.datastructures import UploadFile
     from starlette.requests import Request
 
     from marimo._session.session import Session
@@ -44,40 +48,39 @@ async def parse_request(
 
 @dataclass
 class MultipartRequest(Generic[S]):
-    """Result of parsing a multipart/form-data request body."""
+    """Parsed multipart body. `files` holds un-read `UploadFile` handles
+    so callers can stream large parts instead of buffering."""
 
     body: S
-    files: dict[str, bytes]
+    files: dict[str, UploadFile]
 
 
+@asynccontextmanager
 async def parse_multipart_request(
     request: Request, cls: type[S]
-) -> MultipartRequest[S]:
-    """Parse a multipart/form-data body into a msgspec.Struct + file bytes.
+) -> AsyncIterator[MultipartRequest[S]]:
+    """Parse a multipart/form-data body into a msgspec.Struct + uploads.
 
-    String form fields are validated against `cls`. File upload parts are
-    read fully into memory and returned in `files`, keyed by form-field
-    name (callers look them up explicitly rather than via the struct).
+    Must be used as an async context manager: `UploadFile` parts stay
+    readable for the body of the `async with`, and their spooled temp
+    files are closed on exit.
 
     Raises msgspec.ValidationError if required string fields are missing
     or invalid.
     """
-    # Imported lazily so this module stays import-safe in environments
-    # without starlette (e.g. pyodide).
+    # Lazy import so this module stays import-safe under pyodide.
     from starlette.datastructures import UploadFile
 
-    # Use as an async context manager so any spooled temp files backing
-    # UploadFile parts are closed after parsing.
     async with request.form() as form:
         string_payload: dict[str, Any] = {}
-        files: dict[str, bytes] = {}
+        files: dict[str, UploadFile] = {}
         for key, value in form.multi_items():
             if isinstance(value, UploadFile):
-                files[key] = await value.read()
+                files[key] = value
             elif isinstance(value, str):
                 string_payload[key] = value
-    body = msgspec.convert(string_payload, cls, strict=False)
-    return MultipartRequest(body=body, files=files)
+        body = msgspec.convert(string_payload, cls, strict=False)
+        yield MultipartRequest(body=body, files=files)
 
 
 @runtime_checkable

marimo/_server/files/os_file_system.py

@@ -8,9 +8,10 @@
 import re
 import shutil
 import subprocess
+import tempfile
 from collections import deque
 from pathlib import Path
-from typing import Literal
+from typing import Literal, Protocol
 
 from marimo import _loggers
 from marimo._server.files.file_system import FileSystem
@@ -36,6 +37,29 @@
     "..",
 ]
 
+# 1 MiB. Large enough to amortize syscall overhead, small enough to keep
+# peak memory bounded when streaming.
+_STREAM_CHUNK_SIZE = 1024 * 1024
+
+# Hard cap on streamed uploads. Streaming removes the implicit OOM ceiling
+# that buffered uploads had, so without a cap an authenticated client could
+# exhaust disk. 1 GiB covers normal notebook-data use cases with margin.
+MAX_UPLOAD_BYTES = 1024 * 1024 * 1024
+
+
+class UploadTooLargeError(ValueError):
+    """Raised when a streamed upload exceeds `MAX_UPLOAD_BYTES`.
+
+    Separate type (vs. a bare `ValueError`) so the HTTP layer can map it
+    to a 413 response instead of the generic error path.
+    """
+
+
+class AsyncByteSource(Protocol):
+    """Structural type for things like Starlette's `UploadFile`."""
+
+    async def read(self, size: int = -1, /) -> bytes: ...
+
 
 class OSFileSystem(FileSystem):
     def get_root(self) -> str:
@@ -133,33 +157,32 @@ def open_file(self, path: str, encoding: str | None = None) -> str | bytes:
         except UnicodeDecodeError:
             return file_path.read_bytes()
 
-    def create_file_or_directory(
-        self,
-        path: str,
-        file_type: Literal["file", "directory", "notebook"],
-        name: str,
-        contents: bytes | None,
-    ) -> FileInfo:
+    @staticmethod
+    def _validate_create_name(name: str) -> None:
+        """Reject names that are empty, reserved, or traverse out of the
+        parent. Centralized so HTTP, WASM, and streaming paths all share it.
+        """
         if name in DISALLOWED_NAMES:
             raise ValueError(
                 f"Cannot create file or directory with name {name}"
             )
         if name.strip() == "":
             raise ValueError("Cannot create file or directory with empty name")
-        # Names that traverse out of `path` or escape via separators are
-        # rejected. Validation belongs here (not in the endpoint) so every
-        # caller of OSFileSystem — HTTP, WASM bridge, scripts — is covered.
-        if (
-            "/" in name
-            or "\\" in name
-            or "\x00" in name
-            or name in (".", "..")
-        ):
+        if "/" in name or "\\" in name or "\x00" in name:
             raise ValueError(
                 f"Invalid name {name!r}: must not contain path separators "
                 "or refer to a parent directory"
             )
 
+    def create_file_or_directory(
+        self,
+        path: str,
+        file_type: Literal["file", "directory", "notebook"],
+        name: str,
+        contents: bytes | None,
+    ) -> FileInfo:
+        self._validate_create_name(name)
+
         full_path = Path(path) / name
         full_path = _generate_unique_path(full_path)
 
@@ -192,6 +215,63 @@ def create_file_or_directory(
             ),
         ).file
 
+    async def stream_create_file(
+        self,
+        path: str,
+        name: str,
+        source: AsyncByteSource,
+    ) -> FileInfo:
+        """Stream-write an uploaded file to disk, chunk by chunk.
+
+        Avoids loading the full payload into memory (the HTTP multipart
+        path can otherwise buffer 100 MB at once). Writes to a ``.part``
+        temp file and atomically renames on success so a failed upload
+        doesn't leave a half-written file at the final path.
+        """
+        self._validate_create_name(name)
+
+        parent = Path(path)
+        os.makedirs(parent, exist_ok=True)
+
+        # Atomic O_CREAT|O_EXCL reservation closes the TOCTOU window between
+        # picking a unique name and writing to it.
+        full_path = _claim_unique_path(parent / name)
+
+        tmp = tempfile.NamedTemporaryFile(
+            dir=full_path.parent,
+            prefix=full_path.name + ".",
+            suffix=".part",
+            delete=False,
+        )
+        tmp_path = tmp.name
+        try:
+            # Sync writes are bounded to one chunk between awaits, so event
+            # loop blockage stays small without pulling in aiofiles.
+            written = 0
+            with tmp:
+                while chunk := await source.read(_STREAM_CHUNK_SIZE):
+                    written += len(chunk)
+                    if written > MAX_UPLOAD_BYTES:
+                        raise UploadTooLargeError(
+                            f"Upload exceeds maximum size of "
+                            f"{MAX_UPLOAD_BYTES} bytes"
+                        )
+                    tmp.write(chunk)
+            os.replace(tmp_path, full_path)
+        except BaseException:
+            # Both paths may or may not exist depending on where we failed;
+            # FileNotFoundError on either is expected.
+            for p in (tmp_path, str(full_path)):
+                try:
+                    os.unlink(p)
+                except FileNotFoundError:
+                    pass
+            raise
+
+        # `get_details` would re-read the file (and base64-encode binary),
+        # defeating the streaming win.
+        return self._get_file_info(str(full_path))
+
     def delete_file_or_directory(self, path: str) -> bool:
         if os.path.isdir(path):
             safe_rmtree(path)
@@ -485,6 +565,29 @@ def _generate_unique_path(new_path: str | Path) -> Path:
         i += 1
 
 
+def _claim_unique_path(target: Path) -> Path:
+    """Race-safe variant of `_generate_unique_path`: opens with O_EXCL and
+    walks numbered suffixes on collision. Returns an empty file at the
+    claimed path; the caller writes into or replaces it.
+    """
+    name_without_extension = target.stem
+    extension = target.suffix
+    candidate = target
+    i = 0
+    while True:
+        try:
+            fd = os.open(
+                candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644
+            )
+            os.close(fd)
+            return candidate
+        except FileExistsError:
+            i += 1
+            candidate = target.parent / (
+                f"{name_without_extension}_{i}{extension}"
+            )
+
+
 def _is_allowed_paths(path: str | Path, new_path: str | Path) -> bool:
     file_name = os.path.basename(new_path)
     if file_name in DISALLOWED_NAMES or not file_name.strip():

marimo/_server/models/files.py

@@ -58,16 +58,22 @@ class FileCreateRequest(msgspec.Struct, rename="camel"):
 
 
 class FileCreateMultipartRequest(msgspec.Struct, rename="camel"):
-    """multipart/form-data body for POST /api/files/create."""
+    """multipart/form-data body for POST /api/files/create.
+
+    Schema-only: this struct exists to describe the multipart shape in
+    OpenAPI. At runtime, the endpoint reads the string fields from
+    ``MultipartRequest.body`` and the uploaded bytes from
+    ``MultipartRequest.files["file"]`` — ``body.file`` is never populated.
+    """
 
     # The path where to create the file or directory
     path: str
     # 'file', 'directory', or 'notebook'
     type: FileCreateType
     # The name of the file or directory
     name: str
-    # The raw file bytes (optional). When omitted, an empty file is created
-    # (or, for 'notebook' type, a default notebook template).
+    # OpenAPI-only (see class docstring). `format: binary` makes the
+    # generated spec emit a file-upload schema rather than a base64 string.
     file: Annotated[
         str | None, msgspec.Meta(extra_json_schema={"format": "binary"})
     ] = None

packages/openapi/api.yaml

@@ -1665,7 +1665,11 @@ components:
       title: FileCopyResponse
       type: object
     FileCreateMultipartRequest:
-      description: multipart/form-data body for POST /api/files/create.
+      description: "multipart/form-data body for POST /api/files/create.\n\n    Schema-only:\
+        \ this struct exists to describe the multipart shape in\n    OpenAPI. At runtime,\
+        \ the endpoint reads the string fields from\n    ``MultipartRequest.body``\
+        \ and the uploaded bytes from\n    ``MultipartRequest.files[\"file\"]`` \u2014\
+        \ ``body.file`` is never populated."
       properties:
         file:
           anyOf:

packages/openapi/src/api.ts

@@ -4359,6 +4359,11 @@ export interface components {
     /**
      * FileCreateMultipartRequest
      * @description multipart/form-data body for POST /api/files/create.
+     *
+     *         Schema-only: this struct exists to describe the multipart shape in
+     *         OpenAPI. At runtime, the endpoint reads the string fields from
+     *         ``MultipartRequest.body`` and the uploaded bytes from
+     *         ``MultipartRequest.files["file"]`` — ``body.file`` is never populated.
      */
     FileCreateMultipartRequest: {
       /**

tests/_server/api/endpoints/test_file_explorer.py

@@ -179,6 +179,27 @@ def test_create_rejects_path_traversal_name(client: TestClient) -> None:
     assert not os.path.exists(parent_path)
 
 
+def test_create_returns_413_when_upload_exceeds_size_cap(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Shrink the cap so a small payload trips it.
+    monkeypatch.setattr(
+        "marimo._server.files.os_file_system.MAX_UPLOAD_BYTES", 4
+    )
+    response = client.post(
+        "/api/files/create",
+        headers=HEADERS,
+        data={
+            "path": test_dir,
+            "type": "file",
+            "name": "too_big.bin",
+        },
+        files={"file": ("too_big.bin", b"way too much data")},
+    )
+    assert response.status_code == 413, response.text
+    assert not os.path.exists(os.path.join(test_dir, "too_big.bin"))
+
+
 def test_update_file(client: TestClient) -> None:
     response = client.post(
         "/api/files/update",