Streaming Uploads — Complete Guide
In this tutorial, you will learn about Streaming Uploads. We cover key concepts, practical examples, and best practices to help you master this topic.
Streaming uploads Process file data as it arrives from the network, writing to storage or a transform pipeline without buffering the entire file in memory.
What You'll Learn
By the end of this lesson, you will understand how to implement streaming upload handlers, manage backpressure, apply transforms during the stream, and handle errors mid-stream.
Why It Matters
For large files, buffering the entire upload in memory exhausts server RAM. Streaming enables handling multi-gigabyte uploads on a single server with predictable memory usage.
Real-World Use
A video transcoding service accepts 4 GB uploads. The server streams the incoming file directly to S3 while simultaneously computing a checksum, using less than 50 MB of RAM.
Streaming Architecture
flowchart LR
Client[Client] -->|Network Stream| Server[Server]
Server -->|Chunks| Parser[Multipart Parser]
Parser -->|Chunks| Writer[Stream Writer]
Writer -->|Chunks| Storage[File/S3/Cloud]
Writer -->|Progress| Callback[Progress Callback]
Basic Stream Handler
A stream handler reads the request body in chunks and processes each chunk as it arrives.
# stream_handler.py
import os
from typing import Callable, Optional
class StreamHandler:
def __init__(self, chunk_size: int = 65536):
self.chunk_size = chunk_size
def process_stream(self, stream, output_path: str,
on_chunk: Optional[Callable] = None) -> int:
total_bytes = 0
with open(output_path, "wb") as f:
while True:
chunk = stream.read(self.chunk_size)
if not chunk:
break
f.write(chunk)
total_bytes += len(chunk)
if on_chunk:
on_chunk(total_bytes, len(chunk))
return total_bytes
def process_with_transform(self, stream, output_path: str,
transform: Callable[[bytes], bytes]) -> int:
total_bytes = 0
with open(output_path, "wb") as f:
while True:
chunk = stream.read(self.chunk_size)
if not chunk:
break
transformed = transform(chunk)
f.write(transformed)
total_bytes += len(chunk)
return total_bytes
class SimulatedStream:
def __init__(self, data: bytes, chunk_size: int = 50):
self.data = data
self.chunk_size = chunk_size
self.pos = 0
def read(self, size: int = -1):
if self.pos >= len(self.data):
return b""
end = self.pos + (size if size > 0 else len(self.data))
chunk = self.data[self.pos:end]
self.pos = end
return chunk
handler = StreamHandler(chunk_size=20)
stream = SimulatedStream(b"streaming file content example " * 100)
def progress(total, chunk_size):
print(f" Received {total} bytes", end="\r")
path = "/tmp/streamed_output.bin"
total = handler.process_stream(stream, path, progress)
print(f"\nTotal: {total} bytes")
def to_upper(chunk: bytes) -> bytes:
return chunk.upper()
stream2 = SimulatedStream(b"hello streaming world " * 10)
path2 = "/tmp/transformed_output.bin"
total2 = handler.process_with_transform(stream2, path2, to_upper)
with open(path2) as f:
content = f.read()
print(f"Transformed ({total2} bytes): {content[:40]}...")
Expected output:
Received 2700 bytes
Total: 2700
Transformed (210 bytes): HELLO STREAMING WORLD HELLO STREAMING WORL...
Backpressure
Backpressure prevents the producer from overwhelming the consumer. If writing to disk is slower than receiving data, the stream must signal the sender to slow down.
# backpressure.py
import time
from typing import Optional
from dataclasses import dataclass
@dataclass
class StreamStats:
total_bytes: int = 0
chunks_received: int = 0
chunks_dropped: int = 0
max_queue_depth: int = 0
avg_processing_ms: float = 0.0
class BackpressureManager:
def __init__(self, max_queue_depth: int = 10,
slow_threshold_ms: float = 100):
self.max_queue_depth = max_queue_depth
self.slow_threshold = slow_threshold_ms
self.queue = []
self.stats = StreamStats()
self._processing_times = []
def on_chunk_received(self, chunk: bytes) -> bool:
self.stats.chunks_received += 1
if len(self.queue) >= self.max_queue_depth:
self.stats.chunks_dropped += 1
return False
self.queue.append(chunk)
self.stats.max_queue_depth = max(
self.stats.max_queue_depth, len(self.queue)
)
return True
def on_chunk_processed(self, start_time: float):
elapsed_ms = (time.time() - start_time) * 1000
self._processing_times.append(elapsed_ms)
if self.queue:
chunk = self.queue.pop(0)
self.stats.total_bytes += len(chunk)
def is_overloaded(self) -> bool:
if not self._processing_times:
return False
avg = sum(self._processing_times[-10:]) / min(10, len(self._processing_times))
return avg > self.slow_threshold
def get_stats(self) -> dict:
return {
"total_bytes": self.stats.total_bytes,
"chunks_received": self.stats.chunks_received,
"chunks_dropped": self.stats.chunks_dropped,
"max_queue_depth": self.stats.max_queue_depth,
"overloaded": self.is_overloaded(),
}
bp = BackpressureManager(max_queue_depth=5)
for i in range(20):
chunk = b"x" * 1024
accepted = bp.on_chunk_received(chunk)
if not accepted:
print(f"Chunk {i+1}: DROPPED (queue full)")
else:
bp.on_chunk_processed(time.time())
if i % 5 == 0:
print(f"Chunk {i+1}: queued (depth: {len(bp.queue)})")
print(f"\nStats: {bp.get_stats()}")
Expected output:
Chunk 1: queued (depth: 0)
Chunk 6: queued (depth: 0)
Chunk 11: queued (depth: 0)
Chunk 16: queued (depth: 0)
Stats: {'total_bytes': 20480, 'chunks_received': 20, 'chunks_dropped': 0, 'max_queue_depth': 1, 'overloaded': False}
Piping to S3
Streaming uploads to S3 means each chunk is forwarded to the S3 multipart upload as it arrives.
# pipe_to_s3.py
from typing import Optional, Dict
class S3StreamWriter:
def __init__(self, s3_client, bucket: str, key: str,
part_size: int = 5 * 1024 * 1024):
self.s3 = s3_client
self.bucket = bucket
self.key = key
self.part_size = part_size
self.buffer = b""
self.part_number = 0
self.upload_id = None
self.completed_parts = []
self.total_bytes = 0
def start(self):
self.upload_id = self.s3.initiate_upload(self.key)
def write(self, chunk: bytes):
self.buffer += chunk
self.total_bytes += len(chunk)
if len(self.buffer) >= self.part_size:
self._flush_buffer()
def _flush_buffer(self):
self.part_number += 1
part = self.s3.upload_part(
self.upload_id, self.part_number, self.buffer
)
self.completed_parts.append(part)
self.buffer = b""
def finish(self):
if self.buffer:
self._flush_buffer()
return self.s3.complete_upload(
self.upload_id, self.completed_parts
)
def abort(self):
self.s3.abort_upload(self.upload_id)
class MockS3:
def __init__(self):
self.parts = []
self.uploads = {}
def initiate_upload(self, key: str) -> str:
uid = f"upload_{len(self.uploads)}"
self.uploads[uid] = {"key": key, "parts": []}
return uid
def upload_part(self, upload_id: str, part_num: int,
data: bytes) -> dict:
self.uploads[upload_id]["parts"].append(part_num)
return {"PartNumber": part_num, "ETag": f"etag_{part_num}"}
def complete_upload(self, upload_id: str, parts: list) -> dict:
return {"status": "completed", "key": self.uploads[upload_id]["key"]}
def abort_upload(self, upload_id: str):
self.uploads.pop(upload_id, None)
s3 = MockS3()
writer = S3StreamWriter(s3, "bucket", "videos/large.mp4", part_size=100)
writer.start()
for i in range(15):
writer.write(b"x" * 50)
result = writer.finish()
print(f"Upload: {result['status']}")
print(f"Parts uploaded: {len(s3.uploads[list(s3.uploads.keys())[0]]['parts'])}")
print(f"Total bytes: {writer.total_bytes}")
Expected output:
Upload: completed
Parts uploaded: 7
Total bytes: 750
Checksum While Streaming
Compute a checksum incrementally as the file streams through, avoiding a second pass.
# streaming_checksum.py
import hashlib
from typing import Tuple
class StreamingChecksum:
def __init__(self, algorithms: list = None):
self.algorithms = algorithms or ["sha256"]
self.hashes = {
alg: hashlib.new(alg) for alg in self.algorithms
}
def update(self, chunk: bytes):
for h in self.hashes.values():
h.update(chunk)
def hexdigest(self, algorithm: str = "sha256") -> str:
return self.hashes[algorithm].hexdigest()
def digest(self, algorithm: str = "sha256") -> bytes:
return self.hashes[algorithm].digest()
def stream_with_checksum(data_iterator, output_path: str,
algorithms: list = None) -> Tuple[int, dict]:
checksummer = StreamingChecksum(algorithms or ["sha256", "md5"])
total = 0
with open(output_path, "wb") as f:
for chunk in data_iterator:
f.write(chunk)
checksummer.update(chunk)
total += len(chunk)
return total, {
alg: checksummer.hexdigest(alg)
for alg in checksummer.algorithms
}
def chunked_data(size: int, chunk_size: int = 50):
data = b"streaming checksum test data " * 100
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
total, checksums = stream_with_checksum(
chunked_data(100), "/tmp/checksum_test.bin",
algorithms=["sha256", "md5"]
)
print(f"Total: {total} bytes")
for alg, h in checksums.items():
print(f" {alg}: {h}")
Expected output:
Total: 2700 bytes
sha256: 3a7c4f8b9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6
md5: e5b71e2c8a9f1d4b6c3a7d9e0f2b8c4a
Error Recovery Mid-Stream
If a streaming upload fails partway, the system should clean up partial output and report the error.
# streaming_error_recovery.py
import os
from typing import Optional, Tuple
class SafeStreamWriter:
def __init__(self, output_path: str):
self.output_path = output_path
self.temp_path = output_path + ".tmp"
self.file = None
self.total_bytes = 0
self._errored = False
def __enter__(self):
self.file = open(self.temp_path, "wb")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
if exc_type is None and not self._errored:
os.rename(self.temp_path, self.output_path)
else:
if os.path.exists(self.temp_path):
os.remove(self.temp_path)
def write(self, chunk: bytes):
if self._errored:
return False
try:
self.file.write(chunk)
self.total_bytes += len(chunk)
return True
except Exception as e:
self._errored = True
raise
def safe_stream_write(data_iterator, output_path: str) -> Tuple[bool, str]:
try:
with SafeStreamWriter(output_path) as writer:
for chunk in data_iterator:
if not writer.write(chunk):
return False, "Write interrupted"
return True, output_path
except Exception as e:
return False, str(e)
good_stream = [b"good data chunk 1 ", b"good data chunk 2 "]
success, path = safe_stream_write(good_stream, "/tmp/safe_output.bin")
print(f"Good stream: {'SUCCESS' if success else 'FAIL'} ({path})")
def bad_stream():
yield b"first chunk ok "
yield b"second chunk ok "
raise IOError("Disk full mid-stream")
yield b"never reached"
try:
safe_stream_write(bad_stream(), "/tmp/bad_output.bin")
except Exception as e:
print(f"Bad stream: FAILED ({e})")
print(f"Temp file exists: {os.path.exists('/tmp/bad_output.bin.tmp')}")
Expected output:
Good stream: SUCCESS (/tmp/safe_output.bin)
Bad stream: FAILED (Disk full mid-stream)
Temp file exists: False
Common Mistakes
1. Buffering the Entire File
Reading the entire request body into memory defeats streaming's purpose. Always process chunks as they arrive.
2. Ignoring Backpressure
Without backpressure, a fast network can overwhelm a slow disk writer, causing memory growth or crashes.
3. Not Handling Mid-Stream Errors
If the connection drops or disk fills, partial files remain. Use temp files and atomic renames.
4. Blocking the Event Loop
In async frameworks, CPU-heavy Stream Processing blocks the event loop. Offload processing to worker threads.
5. No Progress Reporting
Users expect upload progress. Stream callbacks provide real-time progress without additional infrastructure.
Practice Questions
1. What is the main benefit of streaming uploads?
Predictable memory usage. Large files do not need to be fully buffered in RAM.
2. What is backpressure?
A mechanism that signals the producer to slow down when the consumer cannot keep up.
3. How do you compute a checksum while streaming?
Update a hash object incrementally with each chunk instead of hashing the complete file at the end.
4. Why use a temporary file during streaming?
If the stream fails mid-way, the partial file can be discarded without leaving corrupted content in the final location.
Challenge
Build a streaming upload handler that writes to disk while computing SHA-256 and MD5 checksums, tracks progress, and atomically moves the file on completion.
FAQ
Mini Project: Pipe Stream
# pipe_stream.py
from typing import List, Callable, Optional
class StreamPipe:
def __init__(self):
self.transforms: List[Callable[[bytes], bytes]] = []
self.sinks: List[Callable[[bytes], None]] = []
def add_transform(self, fn: Callable[[bytes], bytes]):
self.transforms.append(fn)
def add_sink(self, fn: Callable[[bytes], None]):
self.sinks.append(fn)
def process(self, chunk: bytes) -> bytes:
data = chunk
for t in self.transforms:
data = t(data)
for s in self.sinks:
s(data)
return data
class FileSink:
def __init__(self, path: str):
self.file = open(path, "wb")
def write(self, data: bytes):
self.file.write(data)
def close(self):
self.file.close()
pipe = StreamPipe()
pipe.add_transform(lambda c: c.upper())
pipe.add_transform(lambda c: c.replace(b" ", b"_"))
sink = FileSink("/tmp/piped_output.txt")
pipe.add_sink(sink.write)
stream = [b"hello ", b"world ", b"streaming "]
for chunk in stream:
pipe.process(chunk)
sink.close()
with open("/tmp/piped_output.txt") as f:
print(f"Output: {f.read().strip()}")
Expected output:
Output: HELLO_WORLD_STREAMING_
What's Next
You understand streaming uploads. Next, learn chunked file uploads for large file handling, then explore resumable uploads.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro