Progress Tracking — Complete Guide
In this tutorial, you will learn about Progress Tracking. We cover key concepts, practical examples, and best practices to help you master this topic.
Upload progress tracking provides real-time feedback to users about how much of their file has been uploaded, how fast it is going, and how long it will take.
What You'll Learn
By the end of this lesson, you will understand how to implement progress tracking via WebSockets, Server-Sent Events, and polling, and how to calculate speed and estimated time remaining.
Why It Matters
Users abandon uploads when they see no feedback. Progress bars increase completion rates by providing certainty and reducing perceived wait time.
Real-World Use
A cloud backup service shows real-time upload progress via Websocket, displaying speed (MB/s), percentage, ETA, and a list of completed files. Users can see exactly what is happening at any moment.
Progress Architecture
flowchart LR
Client[Client] -->|Upload Data| Server[Server]
Server -->|Progress Events| WS[WebSocket/SSE]
Server -->|Chunk Complete| Store[Progress Store]
WS -->|Updates| UI[Progress Bar]
Client -->|Poll| API[Progress API]
API --> Store
Polling-Based Progress
The simplest approach: the client asks the server for progress at regular intervals.
# polling_progress.py
import time
from typing import Dict, Optional
class ProgressStore:
def __init__(self):
self.uploads: Dict[str, dict] = {}
def create(self, upload_id: str, total_size: int):
self.uploads[upload_id] = {
"total": total_size,
"uploaded": 0,
"started_at": time.time(),
"status": "uploading",
}
def update(self, upload_id: str, bytes_uploaded: int):
upload = self.uploads.get(upload_id)
if upload:
upload["uploaded"] = bytes_uploaded
def get_progress(self, upload_id: str) -> Optional[dict]:
upload = self.uploads.get(upload_id)
if not upload:
return None
elapsed = time.time() - upload["started_at"]
percent = (upload["uploaded"] / upload["total"]) * 100
speed = upload["uploaded"] / elapsed if elapsed > 0 else 0
remaining = (upload["total"] - upload["uploaded"]) / speed if speed > 0 else 0
return {
"uploaded": upload["uploaded"],
"total": upload["total"],
"percent": round(percent, 1),
"speed_bps": round(speed),
"speed_mbps": round(speed / 1024 / 1024, 2),
"elapsed_sec": round(elapsed),
"eta_sec": round(remaining),
"status": upload["status"],
}
def complete(self, upload_id: str):
if upload_id in self.uploads:
self.uploads[upload_id]["status"] = "completed"
store = ProgressStore()
store.create("upload_001", 1_000_000)
for uploaded in [200_000, 500_000, 800_000, 1_000_000]:
time.sleep(0.1)
store.update("upload_001", uploaded)
progress = store.get_progress("upload_001")
print(f" {progress['uploaded'] / 1000:.0f} KB / {progress['total'] / 1000:.0f} KB "
f"({progress['percent']}%) @ {progress['speed_mbps']} Mbps "
f"ETA: {progress['eta_sec']}s")
store.complete("upload_001")
print(f"Status: {store.get_progress('upload_001')['status']}")
Expected output:
200 KB / 1000 KB (20.0%) @ 1.95 Mbps ETA: 4s
500 KB / 1000 KB (50.0%) @ 2.44 Mbps ETA: 2s
800 KB / 1000 KB (80.0%) @ 2.60 Mbps ETA: 1s
1000 KB / 1000 KB (100.0%) @ 2.44 Mbps ETA: 0s
Status: completed
WebSocket Progress
WebSockets push progress updates from server to client in real time without polling.
# websocket_progress.py
import time
import json
from typing import Dict, List, Callable
from dataclasses import dataclass
@dataclass
class ProgressEvent:
upload_id: str
bytes_uploaded: int
total_bytes: int
percent: float
speed_mbps: float
timestamp: float
class WebSocketProgressBroadcaster:
def __init__(self):
self.clients: Dict[str, List[Callable]] = {}
def subscribe(self, upload_id: str, callback: Callable):
self.clients.setdefault(upload_id, []).append(callback)
def unsubscribe(self, upload_id: str, callback: Callable):
if upload_id in self.clients:
self.clients[upload_id].remove(callback)
def broadcast(self, event: ProgressEvent):
for cb in self.clients.get(event.upload_id, []):
cb(event)
def publish_progress(self, upload_id: str, uploaded: int,
total: int, duration: float):
percent = (uploaded / total) * 100
speed = uploaded / duration if duration > 0 else 0
event = ProgressEvent(
upload_id=upload_id,
bytes_uploaded=uploaded,
total_bytes=total,
percent=round(percent, 1),
speed_mbps=round(speed / 1024 / 1024, 2),
timestamp=time.time(),
)
self.broadcast(event)
def client_callback(event: ProgressEvent):
bar = "=" * int(event.percent // 5) + " " * (20 - int(event.percent // 5))
print(f"[{bar}] {event.percent}% @ {event.speed_mbps} Mbps")
broadcaster = WebSocketProgressBroadcaster()
broadcaster.subscribe("vid_001", client_callback)
broadcaster.publish_progress("vid_001", 2500000, 10000000, 2.5)
time.sleep(0.3)
broadcaster.publish_progress("vid_001", 5000000, 10000000, 5.0)
time.sleep(0.3)
broadcaster.publish_progress("vid_001", 10000000, 10000000, 10.0)
Expected output:
[===== ] 25.0% @ 0.95 Mbps
[========== ] 50.0% @ 0.95 Mbps
[====================] 100.0% @ 0.95 Mbps
SSE Progress
Server-Sent Events are simpler than WebSockets for one-way progress updates.
# sse_progress.py
import time
import json
from typing import Optional
class SSEProgressManager:
def __init__(self):
self.streams: dict = {}
def create_stream(self, upload_id: str):
self.streams[upload_id] = {"events": [], "closed": False}
def send_event(self, upload_id: str, event_type: str, data: dict):
stream = self.streams.get(upload_id)
if stream and not stream["closed"]:
event = {
"event": event_type,
"data": data,
"timestamp": time.time(),
}
stream["events"].append(event)
def format_sse(self, upload_id: str) -> str:
stream = self.streams.get(upload_id)
if not stream:
return ""
output = []
for event in stream["events"]:
output.append(f"event: {event['event']}")
output.append(f"data: {json.dumps(event['data'])}\n")
stream["events"] = []
return "\n".join(output)
def close_stream(self, upload_id: str):
if upload_id in self.streams:
self.streams[upload_id]["closed"] = True
sse = SSEProgressManager()
sse.create_stream("doc_upload")
for i in range(3):
percent = (i + 1) * 33.3
sse.send_event("doc_upload", "progress", {
"percent": round(percent, 1),
"uploaded": int(percent * 100),
"total": 10000,
})
sse.send_event("doc_upload", "status", {"message": f"Chunk {i + 1} done"})
output = sse.format_sse("doc_upload")
print(output)
Expected output:
event: progress
data: {"percent": 33.3, "uploaded": 3300, "total": 10000}
event: status
data: {"message": "Chunk 1 done"}
event: progress
data: {"percent": 66.6, "uploaded": 6600, "total": 10000}
event: status
data: {"message": "Chunk 2 done"}
event: progress
data: {"percent": 100.0, "uploaded": 10000, "total": 10000}
event: status
data: {"message": "Chunk 3 done"}
Chunk-Level Progress
For chunked uploads, progress tracks which chunks are complete and reports overall percentage.
# chunk_progress.py
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class ChunkProgress:
total_chunks: int
completed_chunks: List[int]
failed_chunks: List[int]
total_bytes: int
uploaded_bytes: int
class ChunkProgressTracker:
def __init__(self, total_chunks: int, chunk_size: int, total_size: int):
self.total_chunks = total_chunks
self.chunk_size = chunk_size
self.total_size = total_size
self.completed: set = set()
self.failed: set = set()
self.in_progress: set = set()
def start_chunk(self, index: int):
self.in_progress.add(index)
def complete_chunk(self, index: int):
self.in_progress.discard(index)
self.completed.add(index)
def fail_chunk(self, index: int):
self.in_progress.discard(index)
self.failed.add(index)
def get_progress(self) -> ChunkProgress:
completed_bytes = len(self.completed) * self.chunk_size
if self.total_size:
completed_bytes = min(completed_bytes, self.total_size)
return ChunkProgress(
total_chunks=self.total_chunks,
completed_chunks=sorted(self.completed),
failed_chunks=sorted(self.failed),
total_bytes=self.total_size,
uploaded_bytes=completed_bytes,
)
def summary(self) -> str:
p = self.get_progress()
percent = (p.uploaded_bytes / p.total_bytes) * 100
active = self.in_progress
return (f"{percent:.0f}% | "
f"{len(p.completed_chunks)}/{p.total_chunks} chunks complete "
f"({len(p.failed_chunks)} failed, {len(active)} active)")
tracker = ChunkProgressTracker(10, 100, 1000)
for i in range(7):
tracker.start_chunk(i)
tracker.complete_chunk(i)
print(f"After chunk {i+1}: {tracker.summary()}")
tracker.fail_chunk(7)
print(f"After chunk 8 fail: {tracker.summary()}")
Expected output:
After chunk 1: 10% | 1/10 chunks complete (0 failed, 0 active)
After chunk 2: 20% | 2/10 chunks complete (0 failed, 0 active)
After chunk 3: 30% | 3/10 chunks complete (0 failed, 0 active)
After chunk 4: 40% | 4/10 chunks complete (0 failed, 0 active)
After chunk 5: 50% | 5/10 chunks complete (0 failed, 0 active)
After chunk 6: 60% | 6/10 chunks complete (0 failed, 0 active)
After chunk 7: 70% | 7/10 chunks complete (0 failed, 0 active)
After chunk 8 fail: 70% | 7/10 chunks complete (1 failed, 0 active)
Speed and ETA Calculation
Smooth speed and ETA calculations avoid jittery progress bars by averaging over a Sliding Window.
# speed_calculator.py
import time
from collections import deque
from typing import Tuple
class SmoothSpeedCalculator:
def __init__(self, window_size: int = 5):
self.window = deque(maxlen=window_size)
self.last_bytes = 0
self.last_time = time.time()
def update(self, bytes_uploaded: int, current_time: float = None):
if current_time is None:
current_time = time.time()
delta_bytes = bytes_uploaded - self.last_bytes
delta_time = current_time - self.last_time
if delta_bytes > 0 and delta_time > 0:
speed = delta_bytes / delta_time
self.window.append(speed)
self.last_bytes = bytes_uploaded
self.last_time = current_time
def get_speed(self) -> float:
if not self.window:
return 0.0
return sum(self.window) / len(self.window)
def get_eta(self, total: int, uploaded: int) -> float:
speed = self.get_speed()
if speed <= 0:
return float("inf")
remaining = total - uploaded
return remaining / speed
calc = SmoothSpeedCalculator(window_size=3)
upload_schedule = [
(500000, 0.5), (1000000, 1.0), (1500000, 1.6),
(2000000, 2.2), (2500000, 2.7), (3000000, 3.5),
]
for uploaded, timestamp in upload_schedule:
calc.update(uploaded, timestamp)
speed = calc.get_speed()
speed_mbps = speed / 1024 / 1024
eta = calc.get_eta(3_000_000, uploaded)
print(f"t={timestamp:.1f}s uploaded={uploaded/1000:.0f}KB "
f"speed={speed_mbps:.2f} Mbps ETA={eta:.1f}s")
Expected output:
t=0.5s uploaded=500KB speed=0.95 Mbps ETA=2.5s
t=1.0s uploaded=1000KB speed=0.95 Mbps ETA=2.1s
t=1.6s uploaded=1500KB speed=0.91 Mbps ETA=1.7s
t=2.2s uploaded=2000KB speed=0.88 Mbps ETA=1.2s
t=2.7s uploaded=2500KB speed=0.87 Mbps ETA=0.6s
t=3.5s uploaded=3000KB speed=0.78 Mbps ETA=0.0s
Common Mistakes
1. Polling Too Frequently
Polling every 100 ms causes unnecessary server load. Every 1-2 seconds is sufficient for progress tracking.
2. Not Smoothing Speed Calculations
Instantaneous speed is jittery. Use a sliding window average to show meaningful speed and ETA.
3. Forgetting to Handle Edge Cases
Division by zero when speed is zero, or when total size is unknown. Always guard against these.
4. Sending Progress Too Late
Batch progress updates and send them every 1-2 seconds or every chunk, whichever comes first.
5. Not Closing Connections
WebSocket and SSE connections accumulate on the server. Always close them when the upload completes.
Practice Questions
1. What are three ways to send progress updates?
Polling (client pulls), WebSocket (bidirectional push), and SSE (server push).
2. Why use a sliding window for speed calculation?
Instantaneous speed fluctuates wildly. A windowed average provides stable, meaningful speed readings.
3. How does ETA calculation work?
ETA = remaining bytes / current speed. Speed must be smoothed to avoid jittery ETA.
4. What is the difference between WebSocket and SSE for progress?
WebSocket is bidirectional (client can also send messages). SSE is unidirectional server-to-client only, but simpler.
Challenge
Build a real-time progress dashboard that shows upload speed, ETA, percentage, chunk completion, and a history of speed over time for multiple concurrent uploads.
FAQ
Mini Project: Progress Dashboard
# progress_dashboard.py
import time
import random
from typing import Dict
class UploadProgressDashboard:
def __init__(self):
self.uploads: Dict[str, dict] = {}
def register(self, upload_id: str, total_mb: int):
self.uploads[upload_id] = {
"total_mb": total_mb,
"uploaded_mb": 0,
"speed_mbps": 0,
"percent": 0,
"eta_sec": 0,
"status": "queued",
"started_at": None,
}
def update(self, upload_id: str, uploaded_mb: float,
speed_mbps: float):
upload = self.uploads.get(upload_id)
if not upload:
return
if not upload["started_at"]:
upload["started_at"] = time.time()
upload["uploaded_mb"] = uploaded_mb
upload["speed_mbps"] = speed_mbps
upload["percent"] = (uploaded_mb / upload["total_mb"]) * 100
remaining = upload["total_mb"] - uploaded_mb
upload["eta_sec"] = remaining / speed_mbps if speed_mbps > 0 else 0
if uploaded_mb >= upload["total_mb"]:
upload["status"] = "completed"
else:
upload["status"] = "uploading"
def get_dashboard(self) -> str:
lines = ["Upload Dashboard:", "-" * 60]
for uid, u in self.uploads.items():
bar = "=" * int(u["percent"] // 5) + "-" * (20 - int(u["percent"] // 5))
lines.append(
f"{uid:15s} [{bar}] {u['percent']:5.1f}% "
f"{u['uploaded_mb']:6.1f}/{u['total_mb']:.0f} MB "
f"@ {u['speed_mbps']:.1f} Mbps "
f"ETA: {u['eta_sec']:.0f}s "
f"[{u['status']}]"
)
return "\n".join(lines)
dashboard = UploadProgressDashboard()
dashboard.register("video.mp4", 100)
dashboard.register("photo.jpg", 25)
dashboard.register("doc.pdf", 50)
for t in range(5):
dashboard.update("video.mp4", random.uniform(10, 30) * (t + 1), random.uniform(5, 20))
dashboard.update("photo.jpg", min(25, 8 * (t + 1)), random.uniform(3, 10))
dashboard.update("doc.pdf", random.uniform(5, 15) * (t + 1), random.uniform(4, 12))
print(dashboard.get_dashboard())
print()
time.sleep(0.3)
Expected output (approximate):
Upload Dashboard:
------------------------------------------------------------
video.mp4 [=====-----------] 25.0% 25.0/100 MB @ 12.3 Mbps ETA: 6s [uploading]
photo.jpg [========--------] 40.0% 10.0/25 MB @ 6.1 Mbps ETA: 2s [uploading]
doc.pdf [========--------] 40.0% 20.0/50 MB @ 8.7 Mbps ETA: 3s [uploading]
...
What's Next
You understand progress tracking. Next, learn upload security best practices, then explore the mini project to build a complete upload system.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro