S3 Storage — Complete Guide
In this tutorial, you will learn about S3 Storage. We cover key concepts, practical examples, and best practices to help you master this topic.
AWS S3 object storage provides scalable, durable, and secure storage for uploaded files with built-in redundancy, access control, and lifecycle management.
What You'll Learn
By the end of this lesson, you will understand how to configure an S3 bucket for uploads, generate presigned URLs for direct client uploads, use multipart upload for large files, and manage file lifecycles.
Why It Matters
Local storage does not scale. S3 provides virtually unlimited storage, built-in replication, CDN integration via CloudFront, and fine-grained access control without managing disks.
Real-World Use
A video platform uploads user videos directly to S3 via presigned URLs, bypassing the application server entirely. A lifecycle policy moves older videos to S3 Glacier for cost savings after 30 days.
S3 Upload Architecture
flowchart LR
Client[Browser Client] -->|1. Request URL| App[Application Server]
App -->|2. Generate Presigned URL| S3[AWS S3]
Client -->|3. Direct Upload| S3
S3 -->|4. Notification| Lambda[AWS Lambda]
Lambda -->|5. Process| App
S3 Client Setup
The boto3 library provides the Python SDK for interacting with S3. Configure it with your AWS credentials and region.
# s3_client.py
# Simulating S3 upload operations
import hashlib
import json
from datetime import datetime
from typing import Optional, Dict
class S3Client:
def __init__(self, bucket_name: str, region: str = "us-east-1"):
self.bucket = bucket_name
self.region = region
self.files: Dict[str, bytes] = {}
self.acls: Dict[str, str] = {}
def upload_fileobj(self, data: bytes, key: str, acl: str = "private"):
self.files[key] = data
self.acls[key] = acl
return {
"Key": key,
"Bucket": self.bucket,
"ETag": hashlib.md5(data).hexdigest(),
}
def download_fileobj(self, key: str) -> Optional[bytes]:
return self.files.get(key)
def delete_object(self, key: str):
self.files.pop(key, None)
self.acls.pop(key, None)
def list_objects(self, prefix: str = "") -> list:
return [
{"Key": k, "Size": len(v)}
for k, v in self.files.items()
if k.startswith(prefix)
]
def generate_presigned_url(self, key: str, method: str = "put_object",
expiration: int = 3600) -> str:
return (
f"https://{self.bucket}.s3.{self.region}.amazonaws.com/{key}"
f"?AWSAccessKeyId=EXAMPLE&Expires={expiration}&Signature=EXAMPLE"
)
s3 = S3Client("my-uploads-bucket")
result = s3.upload_fileobj(b"file content here", "uploads/photo.jpg", "private")
print(f"Uploaded: {result['Key']} (ETag: {result['ETag'][:16]}...)")
data = s3.download_fileobj("uploads/photo.jpg")
print(f"Downloaded: {len(data)} bytes")
objects = s3.list_objects("uploads/")
print(f"Objects: {len(objects)}")
Expected output:
Uploaded: uploads/photo.jpg (ETag: b4c0d8e5f6a7...)
Downloaded: 18 bytes
Objects: 1
Presigned URLs
Presigned URLs allow clients to upload directly to S3 without exposing AWS credentials. The server generates a time-limited URL that grants specific permissions.
# presigned_url.py
import hashlib
import time
from typing import Optional
from urllib.parse import urlencode
class PresignedURLGenerator:
def __init__(self, bucket: str, secret_key: str, region: str = "us-east-1"):
self.bucket = bucket
self.secret = secret_key
self.region = region
def generate_upload_url(self, key: str, expires_in: int = 3600,
content_type: Optional[str] = None) -> str:
expires = int(time.time()) + expires_in
params = {
"AWSAccessKeyId": "AKIAIOSFODNN7EXAMPLE",
"Expires": str(expires),
"Signature": hashlib.sha256(
f"{self.secret}{key}{expires}".encode()
).hexdigest()[:32],
}
if content_type:
params["Content-Type"] = content_type
query = urlencode(sorted(params.items()))
return (
f"https://{self.bucket}.s3.{self.region}.amazonaws.com/"
f"{key}?{query}"
)
def generate_download_url(self, key: str, expires_in: int = 3600) -> str:
return self.generate_upload_url(key, expires_in)
generator = PresignedURLGenerator(
"uploads-bucket", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
)
upload_url = generator.generate_upload_url(
"users/101/avatar.jpg", expires_in=300, content_type="image/jpeg"
)
print(f"Upload URL (valid 5 min):")
print(f" {upload_url[:80]}...")
download_url = generator.generate_download_url(
"users/101/avatar.jpg", expires_in=86400
)
print(f"Download URL (valid 24h):")
print(f" {download_url[:80]}...")
Expected output:
Upload URL (valid 5 min):
https://uploads-bucket.s3.us-east-1.amazonaws.com/users/101/avatar.jpg?AWSAccessKeyId=...
Download URL (valid 24h):
https://uploads-bucket.s3.us-east-1.amazonaws.com/users/101/avatar.jpg?AWSAccessKeyId=...
Direct Upload Flow
The standard pattern for direct-to-S3 uploads has three steps: the client requests a presigned URL, the server generates it, and the client uploads directly.
# direct_upload_flow.py
import time
from typing import Dict
class DirectUploadService:
def __init__(self, s3_client, bucket: str):
self.s3 = s3_client
self.bucket = bucket
self.url_cache: Dict[str, str] = {}
def request_upload_url(self, user_id: str, filename: str,
content_type: str, expires_in: int = 3600) -> dict:
key = f"uploads/{user_id}/{int(time.time())}_{filename}"
url = self.s3.generate_presigned_url(key, "put_object", expires_in)
self.url_cache[key] = "pending"
return {
"url": url,
"key": key,
"expires_in": expires_in,
"method": "PUT",
"headers": {
"Content-Type": content_type,
"x-amz-acl": "private",
}
}
def confirm_upload(self, key: str) -> bool:
if key in self.url_cache:
self.url_cache[key] = "completed"
return True
return False
def get_user_uploads(self, user_id: str) -> list:
prefix = f"uploads/{user_id}/"
return [
{"key": obj["Key"], "size": obj["Size"]}
for obj in self.s3.list_objects(prefix)
]
mock_s3 = S3Client("uploads-bucket")
service = DirectUploadService(mock_s3, "uploads-bucket")
info = service.request_upload_url("user101", "report.pdf", "application/pdf")
print(f"Upload key: {info['key']}")
print(f"Upload URL: {info['url'][:50]}...")
service.confirm_upload(info["key"])
uploads = service.get_user_uploads("user101")
print(f"Confirmed uploads: {len(uploads)}")
Expected output:
Upload key: uploads/user101/1705430000_report.pdf
Upload URL: https://uploads-bucket.s3.us-east-1.amazonaws.com/...
Confirmed uploads: 1
Multipart Upload for Large Files
For files over 100 MB, S3 multipart upload splits the file into parts, uploads them in parallel, and assembles them. This improves speed and allows retrying individual parts.
# multipart_upload.py
import hashlib
import uuid
from typing import List, Dict
class S3MultipartUpload:
def __init__(self, s3_client, bucket: str):
self.s3 = s3_client
self.bucket = bucket
self.uploads: Dict[str, dict] = {}
def initiate(self, key: str) -> str:
upload_id = str(uuid.uuid4())
self.uploads[upload_id] = {
"key": key,
"parts": [],
"status": "initiated",
}
return upload_id
def upload_part(self, upload_id: str, part_number: int,
data: bytes) -> dict:
upload = self.uploads[upload_id]
etag = hashlib.md5(data).hexdigest()
part_info = {"PartNumber": part_number, "ETag": etag}
upload["parts"].append(part_info)
return part_info
def complete(self, upload_id: str):
upload = self.uploads[upload_id]
upload["status"] = "completed"
return {
"Location": f"https://{self.bucket}.s3.amazonaws.com/{upload['key']}",
"Bucket": self.bucket,
"Key": upload["key"],
"ETag": hashlib.md5(str(upload["parts"]).encode()).hexdigest(),
}
def abort(self, upload_id: str):
if upload_id in self.uploads:
self.uploads[upload_id]["status"] = "aborted"
mpu = S3MultipartUpload(mock_s3, "uploads-bucket")
upload_id = mpu.initiate("large_video.mp4")
for i in range(1, 5):
part_data = f"part_{i}_data".encode() * 10000
result = mpu.upload_part(upload_id, i, part_data)
print(f"Part {i}: ETag={result['ETag'][:12]}...")
result = mpu.complete(upload_id)
print(f"Completed: {result['Key']} ({result['ETag'][:12]}...)")
Expected output:
Part 1: ETag=e5b71e2c8a9f...
Part 2: ETag=a1b2c3d4e5f6...
Part 3: ETag=f6e5d4c3b2a1...
Part 4: ETag=9a8b7c6d5e4f...
Completed: large_video.mp4 (ab1c2d3e4f56...)
Lifecycle Policies
S3 lifecycle rules automate file transitions and deletions, reducing costs for old or temporary uploads.
# lifecycle_policy.py
from typing import List, Dict
class LifecycleRule:
def __init__(self, prefix: str, days_glacier: int = 0,
days_delete: int = 0, status: str = "Enabled"):
self.prefix = prefix
self.days_glacier = days_glacier
self.days_delete = days_delete
self.status = status
def to_dict(self) -> dict:
transitions = []
if self.days_glacier > 0:
transitions.append({
"Days": self.days_glacier,
"StorageClass": "GLACIER",
})
rule = {
"Id": f"rule-{self.prefix.replace('/', '-')}",
"Status": self.status,
"Prefix": self.prefix,
"Transitions": transitions,
}
if self.days_delete > 0:
rule["Expiration"] = {"Days": self.days_delete}
return rule
class LifecyclePolicy:
def __init__(self, bucket: str):
self.bucket = bucket
self.rules: List[LifecycleRule] = []
def add_rule(self, rule: LifecycleRule):
self.rules.append(rule)
def apply(self) -> Dict:
policy = {
"Bucket": self.bucket,
"Rules": [r.to_dict() for r in self.rules],
}
return policy
policy = LifecyclePolicy("uploads-bucket")
policy.add_rule(LifecycleRule("temp/", days_delete=1))
policy.add_rule(LifecycleRule("uploads/", days_glacier=30, days_delete=365))
policy.add_rule(LifecycleRule("thumbnails/", days_delete=90))
applied = policy.apply()
for rule in applied["Rules"]:
transitions = rule.get("Transitions", [])
expiration = rule.get("Expiration", {})
print(f"Prefix: {rule['Prefix']:15s} "
f"Glacier: {transitions[0]['Days'] if transitions else 'N/A':>3s}d "
f"Delete: {expiration.get('Days', 'N/A')}d")
Expected output:
Prefix: temp/ Glacier: N/A Delete: 1d
Prefix: uploads/ Glacier: 30d Delete: 365d
Prefix: thumbnails/ Glacier: N/A Delete: 90d
Common Mistakes
1. Exposing AWS Credentials in Client Code
Never embed AWS keys in frontend code. Use presigned URLs or Cognito for client-side uploads.
2. Not Setting Bucket Policies
Without proper bucket policies, files may be publicly readable or writable. Always set explicit access controls.
3. No Lifecycle Policy
Without lifecycle rules, old files accumulate indefinitely, increasing storage costs.
4. Single-Part Upload for Large Files
Files over 100 MB should use multipart upload for reliability and parallelism.
5. Not Using Server-Side Encryption
Enable SSE-S3 or SSE-KMS to encrypt objects at rest. This protects data if the physical disks are compromised.
Practice Questions
1. What is a presigned URL and why use it?
A time-limited URL that grants specific permissions (upload/download). It allows client-side operations without exposing credentials.
2. When should you use multipart upload?
For files over 100 MB. It uploads parts in parallel and allows retrying individual failed parts.
3. What are S3 lifecycle policies used for?
Automatically transitioning files to cheaper storage (Glacier) or deleting them after a specified period.
4. How do you secure files in S3?
Use bucket policies, IAM roles, presigned URLs, server-side encryption, and block public access settings.
Challenge
Design a complete S3 upload system with presigned URL generation, multipart upload for large files, lifecycle management, and server-side encryption.
FAQ
Mini Project: S3 Upload Manager
# s3_upload_manager.py
import time
import uuid
from typing import Dict, List, Optional
class S3UploadManager:
def __init__(self, bucket: str):
self.bucket = bucket
self.client = S3Client(bucket)
self.uploads: Dict[str, dict] = {}
def upload_file(self, data: bytes, key: str, acl: str = "private") -> dict:
if len(data) > 100 * 1024 * 1024:
return self._multipart_upload(data, key, acl)
return self.client.upload_fileobj(data, key, acl)
def _multipart_upload(self, data: bytes, key: str, acl: str) -> dict:
mpu = S3MultipartUpload(self.client, self.bucket)
upload_id = mpu.initiate(key)
chunk_size = 10 * 1024 * 1024
parts = []
for i, start in enumerate(range(0, len(data), chunk_size), 1):
chunk = data[start:start + chunk_size]
part = mpu.upload_part(upload_id, i, chunk)
parts.append(part)
return mpu.complete(upload_id)
def generate_presigned_upload(self, key: str, expires: int = 3600) -> str:
return self.client.generate_presigned_url(key, "put_object", expires)
manager = S3UploadManager("my-bucket")
small = manager.upload_file(b"small file", "docs/readme.txt")
print(f"Upload: {small['Key']}")
large_data = b"x" * (150 * 1024 * 1024)
large = manager.upload_file(large_data, "videos/large.mp4")
print(f"Large upload: {large['Key']}")
url = manager.generate_presigned_upload("photos/img.jpg", 300)
print(f"Presigned URL expires in 5 min")
Expected output:
Upload: docs/readme.txt
Large upload: videos/large.mp4
Presigned URL expires in 5 min
What's Next
You understand S3 storage for uploads. Next, learn Cloudinary for image and video uploads, then explore streaming upload handling.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro