import os
import uuid
from flask import Blueprint, current_app, jsonify, request
from werkzeug.exceptions import RequestEntityTooLarge
from .streaming import make_stream_factory, DiskStreamTarget
from .safe_names import SafeNames
bp = Blueprint("uploads", __name__)
MAX_FILE_BYTES = 5 * 1024 * 1024 * 1024 # 5 GiB per part
@bp.route("/uploads", methods=["POST"])
def stream_upload():
root = current_app.config["UPLOAD_ROOT"]
names = SafeNames(root)
upload_id = uuid.uuid4().hex
written = []
factory = make_stream_factory(names, upload_id, MAX_FILE_BYTES, written)
try:
# Cap the whole body; parse_form_data streams each part via the factory.
request.max_content_length = MAX_FILE_BYTES * 4
_, files = request.parse_form_data(
request.environ, stream_factory=factory
)
except RequestEntityTooLarge:
_cleanup(written)
return jsonify(error="file too large"), 413
except Exception:
_cleanup(written)
raise
saved = [
{"field": key, "path": f.stream.path, "bytes": f.stream.bytes_written}
for key, f in files.items(multi=True)
if isinstance(f.stream, DiskStreamTarget)
]
return jsonify(upload_id=upload_id, files=saved), 201
def _cleanup(paths):
for path in paths:
try:
os.unlink(path)
except OSError:
pass
import os
from werkzeug.exceptions import RequestEntityTooLarge
class DiskStreamTarget:
def __init__(self, path, max_bytes):
self.path = path
self.max_bytes = max_bytes
self.bytes_written = 0
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
self._fd = os.open(path, flags, 0o640)
def write(self, chunk):
self.bytes_written += len(chunk)
if self.bytes_written > self.max_bytes:
raise RequestEntityTooLarge()
return os.write(self._fd, chunk)
def seek(self, offset, whence=os.SEEK_SET):
return os.lseek(self._fd, offset, whence)
def tell(self):
return os.lseek(self._fd, 0, os.SEEK_CUR)
def close(self):
try:
os.fsync(self._fd)
finally:
os.close(self._fd)
def make_stream_factory(names, upload_id, max_bytes, written):
counter = {"n": 0}
def stream_factory(total_content_length, content_type, filename, content_length=None, start=0):
counter["n"] += 1
part_name = filename or ("part-%d" % counter["n"])
path = names.secure_join(upload_id, part_name)
os.makedirs(os.path.dirname(path), exist_ok=True)
target = DiskStreamTarget(path, max_bytes)
written.append(path)
return target
return stream_factory
import os
from werkzeug.utils import secure_filename
from werkzeug.exceptions import BadRequest
class SafeNames:
def __init__(self, root):
self.root = os.path.realpath(root)
def secure_join(self, upload_id, filename):
name = secure_filename(filename)
if not name:
name = "unnamed"
candidate = os.path.realpath(
os.path.join(self.root, upload_id, name)
)
expected_prefix = os.path.join(self.root, upload_id)
if not candidate.startswith(os.path.realpath(expected_prefix) + os.sep):
raise BadRequest("invalid upload path")
return candidate
By default, Flask (via Werkzeug) buffers an incoming multipart/form-data body: small files land in memory and larger ones spill into a temporary file that is then re-read by application code. For big uploads this doubles disk I/O and can pin large amounts of memory. This snippet shows how to intercept the parse so each field is streamed straight to its final destination, giving constant memory use regardless of upload size.
The stream_upload endpoint disables Werkzeug's automatic form parsing by setting request.max_content_length for a hard cap and calling request.parse_form_data with a custom stream_factory. Werkzeug invokes that factory once per file part and hands back a writable stream that it feeds in chunks as the socket delivers them. The factory returned by make_stream_factory opens the target path with os.open using O_CREAT | O_EXCL so a duplicate upload_id fails loudly instead of clobbering data, then wraps the fd so the caller can .write() directly to disk.
DiskStreamTarget is the thin wrapper Werkzeug writes into. It tracks bytes_written and enforces max_bytes inline, raising RequestEntityTooLarge mid-stream the moment a part exceeds the limit — important because relying on Content-Length alone lets a lying client exhaust disk. Because the wrapper implements write, seek, tell, and close, Werkzeug treats it like a normal file object and never allocates a SpooledTemporaryFile.
The SafeNames helper centralizes path safety: secure_join runs the client filename through secure_filename and resolves the result under a fixed root, rejecting any path that escapes it. This defends against directory-traversal payloads like ../../etc/passwd that arrive in the filename header of a part.
The key trade-off is that a streamed part is committed to disk as it arrives, so a validation failure after the fact requires an explicit cleanup — the endpoint deletes partial files in an except block. This pattern suits ingestion services, media pipelines, or any endpoint accepting multi-gigabyte uploads where buffering the whole body is not viable. For small forms the default parser is simpler and preferable.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
# Installation
# rails active_storage:install
# rails db:migrate
# config/storage.yml
local:ActiveStorage for file uploads and attachments
package pools
import (
"bytes"
"sync"
)sync.Pool for bytes.Buffer to reduce allocations in hot paths
import type { IncomingMessage, ServerResponse } from "http";
const MIN_BYTES = 1024;
const INCOMPRESSIBLE = /^(image|video|audio)\/|application\/(zip|gzip|x-brotli|pdf|octet-stream)/i;Response compression (only when it helps)
class Tag < ApplicationRecord
has_many :taggings, dependent: :destroy
scope :top, ->(limit = 20) {
joins(:taggings)
.group(Arel.sql("tags.id"))Memory-Safe “top tags” aggregation with pluck + group
Share this code
Here's the card — post it anywhere.
