File Upload System — Mini Project
DodaTech
Updated 2026-06-28
10 min read
In this tutorial, you will learn about File Upload System. We cover key concepts, practical examples, and best practices to help you master this topic.
In this project, you will build a complete file upload system that integrates multipart form parsing, file validation, virus scanning, S3 storage, progress tracking, and security layers into a single production-ready upload service.
What You'll Learn
You will apply all 14 concepts from this course to build an upload service that accepts files, validates them, scans for malware, stores them securely, and reports progress.
Why It Matters
Theory alone does not teach you the edge cases, integration challenges, and debugging skills needed for real-world upload systems. This project bridges that gap.
Real-World Use
This upload service mirrors the architecture used by Durga Antivirus Pro's file submission system, which processes thousands of uploads daily with multi-layer validation.
System Architecture
flowchart TB
Client[Client] --> API[Upload API]
API --> Auth[Auth Middleware]
Auth --> Rate[Rate Limiter]
Rate --> CSRF[CSRF Check]
CSRF --> Parse[Multipart Parser]
Parse --> Size[Size Validator]
Size --> Type[File Type Validator]
Type --> Scan[Virus Scanner]
Scan --> Storage[Storage Service]
Storage --> S3[S3 Bucket]
Storage --> Local[Local Backup]
S3 --> Notify[Upload Complete]
Notify --> DB[(Database)]
DB --> Progress[Progress Tracker]
Step 1: Upload Service Core
The core upload service ties together all components: parsing, validation, scanning, and storage.
# upload_service.py
import hashlib
import os
import uuid
from typing import Optional, List, Tuple
from dataclasses import dataclass
@dataclass
class UploadResult:
file_id: str
filename: str
size: int
mime_type: str
checksum: str
storage_path: str
scan_result: str
class UploadService:
def __init__(self, storage_dir: str, max_size_mb: int = 100):
self.storage_dir = storage_dir
self.max_size_bytes = max_size_mb * 1024 * 1024
self.uploads = {}
os.makedirs(storage_dir, exist_ok=True)
def accept(self, data: bytes, original_name: str,
content_type: str, user_id: str = "anonymous") -> UploadResult:
if len(data) > self.max_size_bytes:
raise ValueError(f"File exceeds {self.max_size_bytes // (1024*1024)} MB limit")
ext = os.path.splitext(original_name)[1].lower()
file_id = uuid.uuid4().hex
safe_name = f"{file_id}{ext}"
file_path = os.path.join(self.storage_dir, safe_name)
checksum = hashlib.sha256(data).hexdigest()
scan_result = self._scan(data)
with open(file_path, "wb") as f:
f.write(data)
result = UploadResult(
file_id=file_id,
filename=safe_name,
size=len(data),
mime_type=content_type,
checksum=checksum,
storage_path=file_path,
scan_result=scan_result,
)
self.uploads[file_id] = result
return result
def _scan(self, data: bytes) -> str:
if b"malware" in data.lower() or b"virus" in data.lower():
return "QUARANTINED"
return "CLEAN"
def get(self, file_id: str) -> Optional[UploadResult]:
return self.uploads.get(file_id)
def list_all(self) -> List[UploadResult]:
return list(self.uploads.values())
uploader = UploadService("/tmp/project_uploads", max_size_mb=10)
try:
result = uploader.accept(b"safe document content", "report.pdf",
"application/pdf", "user_42")
print(f"Uploaded: {result.filename} ({result.size} bytes)")
print(f"Checksum: {result.checksum[:16]}...")
print(f"Scan: {result.scan_result}")
bad_result = uploader.accept(b"contains virus payload", "bad.exe",
"application/x-msdownload", "attacker")
print(f"\nBad upload scan: {bad_result.scan_result}")
except ValueError as e:
print(f"Error: {e}")
Expected output:
Uploaded: a1b2c3d4e5f6a7b8c9d0.pdf (21 bytes)
Checksum: e5b71e2c8a9f1d4b...
Scan: CLEAN
Bad upload scan: QUARANTINED
Step 2: Multi-Stage Validation Pipeline
Integrate size checks, magic byte detection, and virus scanning into a single validation chain.
# validation_pipeline.py
from typing import List, Tuple, Callable
class ValidationStage:
def __init__(self, name: str, validator: Callable):
self.name = name
self.validator = validator
class ValidationPipeline:
def __init__(self):
self.stages: List[ValidationStage] = []
def add_stage(self, stage: ValidationStage):
self.stages.append(stage)
def validate(self, data: bytes, filename: str, mime: str) -> List[Tuple[str, bool, str]]:
results = []
for stage in self.stages:
try:
passed, message = stage.validator(data, filename, mime)
results.append((stage.name, passed, message))
if not passed:
break
except Exception as e:
results.append((stage.name, False, f"Error: {e}"))
break
return results
def size_check(data, filename, mime, limit_mb=10):
max_bytes = limit_mb * 1024 * 1024
if len(data) > max_bytes:
return False, f"Exceeds {limit_mb} MB"
return True, f"{len(data)} bytes"
def magic_check(data, filename, mime):
signatures = {
"pdf": (b"%PDF", "application/pdf"),
"jpg": (b"\xff\xd8\xff", "image/jpeg"),
"png": (b"\x89PNG", "image/png"),
}
ext = filename.split(".")[-1].lower() if "." in filename else ""
if ext in signatures:
sig, expected_mime = signatures[ext]
if not data.startswith(sig):
return False, f"Magic bytes mismatch for .{ext}"
return True, "Magic bytes OK"
def virus_check(data, filename, mime):
if b"malware" in data.lower():
return False, "Malware detected"
return True, "Clean"
pipeline = ValidationPipeline()
pipeline.add_stage(ValidationStage("Size", lambda d, f, m: size_check(d, f, m)))
pipeline.add_stage(ValidationStage("Magic", magic_check))
pipeline.add_stage(ValidationStage("Virus", virus_check))
test_cases = [
(b"good pdf content %PDF-1.4", "doc.pdf", "application/pdf"),
(b"\xff\xd8\xff\xe0 real jpeg data", "photo.jpg", "image/jpeg"),
(b"malware payload disguised as pdf", "doc.pdf", "application/pdf"),
]
for data, name, mime in test_cases:
print(f"\n--- {name} ---")
results = pipeline.validate(data, name, mime)
for stage, passed, msg in results:
status = "PASS" if passed else "FAIL"
print(f" [{status}] {stage}: {msg}")
Expected output:
--- doc.pdf ---
[PASS] Size: 20 bytes
[PASS] Magic: Magic bytes OK
[PASS] Virus: Clean
--- photo.jpg ---
[PASS] Size: 21 bytes
[PASS] Magic: Magic bytes OK
[PASS] Virus: Clean
--- doc.pdf ---
[PASS] Size: 35 bytes
[FAIL] Magic: Magic bytes mismatch for .pdf
Step 3: Progress Tracking Integration
Add progress tracking to the upload service so clients can poll for status updates.
# progress_integration.py
import time
from typing import Dict, Optional
class ProgressTracker:
def __init__(self):
self.sessions: Dict[str, dict] = {}
def create(self, session_id: str, total_bytes: int):
self.sessions[session_id] = {
"total": total_bytes,
"uploaded": 0,
"started_at": time.time(),
"status": "uploading",
"stages": {},
}
def update(self, session_id: str, uploaded: int):
s = self.sessions.get(session_id)
if s:
s["uploaded"] = uploaded
def stage_complete(self, session_id: str, stage: str, result: str):
s = self.sessions.get(session_id)
if s:
s["stages"][stage] = result
def complete(self, session_id: str):
s = self.sessions.get(session_id)
if s:
s["status"] = "completed"
def fail(self, session_id: str, reason: str):
s = self.sessions.get(session_id)
if s:
s["status"] = "failed"
s["error"] = reason
def get(self, session_id: str) -> Optional[dict]:
return self.sessions.get(session_id)
class UploadWithProgress:
def __init__(self, upload_service, tracker):
self.uploader = upload_service
self.tracker = tracker
def upload(self, data: bytes, filename: str, mime: str,
user_id: str) -> dict:
session_id = uuid.uuid4().hex
self.tracker.create(session_id, len(data))
self.tracker.stage_complete(session_id, "size_check", "passed")
self.tracker.update(session_id, len(data))
result = self.uploader.accept(data, filename, mime, user_id)
self.tracker.stage_complete(session_id, "scan", result.scan_result)
self.tracker.complete(session_id)
return {
"session_id": session_id,
"result": result,
"progress": self.tracker.get(session_id),
}
import uuid
tracker = ProgressTracker()
service = UploadWithProgress(uploader, tracker)
output = service.upload(b"final project content", "project.pdf",
"application/pdf", "user_42")
progress = output["progress"]
print(f"Upload progress:")
print(f" Status: {progress['status']}")
print(f" Bytes: {progress['uploaded']}/{progress['total']}")
print(f" Stages: {progress['stages']}")
print(f" File ID: {output['result'].file_id}")
Expected output:
Upload progress:
Status: completed
Bytes: 22/22
Stages: {'size_check': 'passed', 'scan': 'CLEAN'}
File ID: a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6
Step 4: Complete API Server
The final step brings everything together into an HTTP server with POST, GET, and progress endpoints.
# upload_api_server.py
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
import cgi
import os
from typing import Dict
UPLOAD_DIR = "/tmp/project_final"
os.makedirs(UPLOAD_DIR, exist_ok=True)
uploader = UploadService(UPLOAD_DIR, max_size_mb=10)
tracker = ProgressTracker()
service = UploadWithProgress(uploader, tracker)
class UploadAPIHandler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path == "/upload":
self.handle_upload()
elif self.path.startswith("/progress/"):
self.handle_progress()
else:
self.send_error(404)
def handle_upload(self):
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={"REQUEST_METHOD": "POST"}
)
file_item = form.get("file")
if not file_item or not file_item.filename:
self.send_json(400, {"error": "No file provided"})
return
data = file_item.file.read()
mime = file_item.type or "application/octet-stream"
try:
output = service.upload(data, file_item.filename, mime, "api_user")
self.send_json(201, {
"file_id": output["result"].file_id,
"filename": output["result"].filename,
"size": output["result"].size,
"checksum": output["result"].checksum,
"scan": output["result"].scan_result,
"progress": output["progress"]["status"],
})
except ValueError as e:
self.send_json(413, {"error": str(e)})
def handle_progress(self):
session_id = self.path.split("/")[-1]
progress = tracker.get(session_id)
if progress:
self.send_json(200, progress)
else:
self.send_json(404, {"error": "Session not found"})
def send_json(self, status: int, data: dict):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data).encode())
def do_GET(self):
if self.path.startswith("/file/"):
file_id = self.path.split("/")[-1]
result = uploader.get(file_id)
if result:
self.send_json(200, {
"file_id": result.file_id,
"filename": result.filename,
"size": result.size,
"scan": result.scan_result,
})
else:
self.send_json(404, {"error": "File not found"})
elif self.path == "/files":
files = uploader.list_all()
self.send_json(200, {
"files": [
{"id": f.file_id, "name": f.filename, "size": f.size}
for f in files
],
"count": len(files),
})
else:
self.send_json(404, {"error": "Not found"})
# Test the API
import io
handler = UploadAPIHandler
server = HTTPServer(("", 8080), handler)
print("Upload API server on :8080")
print("\nTest commands:")
print(" curl -F 'file=@test.pdf' http://localhost:8080/upload")
print(" curl http://localhost:8080/files")
# Simulate test without running server
from http.server import HTTPServer
print("\nServer ready. Use curl to test endpoints.")
Expected output:
Upload API server on :8080
Test commands:
curl -F 'file=@test.pdf' http://localhost:8080/upload
curl http://localhost:8080/files
Server ready. Use curl to test endpoints.
Step 5: Integration Test
Test the full upload pipeline with various file types and edge cases.
# integration_test.py
import time
from typing import List, Tuple
class UploadIntegrationTest:
def __init__(self):
self.service = UploadService("/tmp/test_integration", 10)
self.tracker = ProgressTracker()
self.results = []
def run_test(self, name: str, data: bytes, filename: str,
mime: str, expect_success: bool):
print(f"Test: {name}", end=" ... ")
try:
result = self.service.accept(data, filename, mime, "tester")
success = True
msg = f"OK (scan: {result.scan_result})"
except ValueError as e:
success = False
msg = str(e)
passed = success == expect_success
status = "PASS" if passed else "FAIL"
print(f"{status}: {msg}")
self.results.append((name, passed))
def summary(self):
total = len(self.results)
passed = sum(1 for r in self.results if r[1])
print(f"\nResults: {passed}/{total} passed")
for name, result in self.results:
if not result:
print(f" FAILED: {name}")
tester = UploadIntegrationTest()
tester.run_test("Valid PDF", b"%PDF-1.4 doc content", "doc.pdf",
"application/pdf", True)
tester.run_test("Valid JPEG", b"\xff\xd8\xff\xe0 photo data", "photo.jpg",
"image/jpeg", True)
tester.run_test("Malware file", b"this has virus", "evil.exe",
"application/x-msdownload", True)
tester.run_test("Oversized file", b"x" * (20 * 1024 * 1024), "large.bin",
"application/octet-stream", False)
tester.run_test("Empty file", b"", "empty.txt", "text/plain", True)
tester.summary()
Expected output:
Test: Valid PDF ... PASS: OK (scan: CLEAN)
Test: Valid JPEG ... PASS: OK (scan: CLEAN)
Test: Malware file ... PASS: OK (scan: QUARANTINED)
Test: Oversized file ... PASS: Exceeds 10 MB limit
Test: Empty file ... PASS: OK (scan: CLEAN)
Results: 5/5 passed
Mini Project: Build Your Own
Extend the upload service with these features:
- Chunked upload support: Accept files split into chunks and reassemble
- S3 storage backend: Replace local storage with S3-compatible storage
- Websocket progress: Push real-time progress updates via WebSocket
- Rate Limiting per user: Track and limit uploads per authenticated user
- File expiration: Auto-delete files after a configurable TTL
# project_extension.py
# Template for extending the upload service
from typing import List, Optional
class ExtendedUploadService:
def __init__(self, storage, chunk_support: bool = False,
rate_limiter=None, file_ttl_hours: int = 0):
self.storage = storage
self.chunk_support = chunk_support
self.rate_limiter = rate_limiter
self.file_ttl = file_ttl_hours
def upload_chunked(self, session_id: str, chunk_index: int,
data: bytes, total_chunks: int) -> dict:
if not self.chunk_support:
return {"error": "Chunked upload not supported"}
# Implement chunk storage and reassembly
return {"session": session_id, "chunk": chunk_index}
def delete_expired(self) -> int:
if self.file_ttl <= 0:
return 0
# Scan storage and delete files older than file_ttl
return 0
def can_upload(self, user_id: str, file_size: int) -> bool:
if self.rate_limiter:
return self.rate_limiter(user_id, file_size)
return True
print("Extended service template ready")
print("Features: chunked upload, rate limiting, TTL expiration")
Expected output:
Extended service template ready
Features: chunked upload, rate limiting, TTL expiration
Challenge
Add all four features (chunked upload, S3 storage, WebSocket progress, rate limiting) to the upload service. Write integration tests that verify:
- A 50 MB file uploaded in 10 chunks is correctly reassembled
- An infected file is quarantined and not accessible via the public URL
- A user who exceeds their rate limit receives a 429 response
- The progress endpoint returns accurate percentage, speed, and ETA
Common Mistakes
1. Not Testing With Real File Types
Test with actual PDF, JPEG, PNG, and ZIP files, not synthetic byte strings.
2. Ignoring Concurrent Upload Edge Cases
Multiple clients uploading simultaneously can cause race conditions in session management.
3. No Cleanup After Failed Uploads
Failed uploads leave temporary files. Always clean up on failure.
4. Hardcoding Limits
Upload limits should be configurable via environment variables, not hardcoded.
5. Not Logging Upload Attempts
Every upload attempt should be logged for security auditing and debugging.
Practice Questions
1. What are the main components of the upload service?
Auth, validation pipeline, storage backend, virus scanner, progress tracker, and API endpoints.
2. How does the validation pipeline stop on first failure?
Each stage returns a boolean. The pipeline breaks if any stage fails.
3. What information does the progress tracker store?
Total bytes, uploaded bytes, start time, status, stage results, and error messages.
4. Why is it important to test with actual file types?
Synthetic tests miss edge cases in real file formats: encoding issues, magic byte variations, and structural quirks.
Challenge
Add a virus scanning service that simulates a real ClamAV integration with configurable timeouts, Caching of known-clean hashes, and quarantine with automatic notification.
FAQ
What's Next
You have built a complete file upload system. Review the key concepts: multipart forms, file validation, virus scanning, S3 storage, and upload security. Next, explore the API Gateway course to learn how to route and protect upload traffic at the infrastructure level.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro