FastAPI Request Files
File uploads are a common requirement in web applications. Whether you're building an image sharing platform, document management system, or any application that needs to accept user files, FastAPI provides powerful tools to handle file uploads efficiently. In this guide, we'll explore how to work with file uploads in FastAPI applications.
Introduction to File Handling in FastAPI
FastAPI makes it easy to handle file uploads through its integration with Python's standard libraries and the python-multipart
package. With FastAPI, you can:
- Accept single or multiple files
- Process uploaded files in various formats
- Validate file types, sizes, and contents
- Save files to the filesystem or cloud storage
- Stream large files efficiently
Let's start by setting up our environment and understanding the basics of file handling in FastAPI.
Prerequisites
Before we begin, make sure you have FastAPI and its dependencies installed:
pip install fastapi uvicorn python-multipart
The python-multipart
package is essential for parsing form data that includes files.
Basic File Upload
Single File Upload
Let's start with a simple example of handling a single file upload:
from fastapi import FastAPI, File, UploadFile
import shutil
from pathlib import Path
from tempfile import NamedTemporaryFile
app = FastAPI()
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(await file.read())
}
In this example:
- We use the
UploadFile
class, which offers several advantages over standard bytes:- It uses a spooled file, so large files won't consume all memory
- You can access metadata like filename and content-type
- It has built-in async methods
When you run this endpoint and send a file, you'll get a response like:
{
"filename": "example.jpg",
"content_type": "image/jpeg",
"size": 123456
}
Saving Uploaded Files
To save an uploaded file to the server's filesystem:
@app.post("/upload-and-save/")
async def upload_and_save(file: UploadFile = File(...)):
# Create uploads directory if it doesn't exist
upload_dir = Path("uploads")
upload_dir.mkdir(exist_ok=True)
# Save the file
destination = upload_dir / file.filename
with destination.open("wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {"filename": file.filename, "saved_path": str(destination)}
This saves the uploaded file to an "uploads" directory in your project.
Multiple File Uploads
FastAPI can handle multiple file uploads as well:
@app.post("/upload-multiple/")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
file_info = []
for file in files:
file_info.append({
"filename": file.filename,
"content_type": file.content_type,
"size": len(await file.read())
})
return {"files": file_info}
Advanced File Handling
File Validation
You often need to validate uploaded files to ensure they meet certain criteria:
from fastapi import HTTPException
@app.post("/upload-image/")
async def upload_image(file: UploadFile = File(...)):
# Check if file is an image
if not file.content_type.startswith("image/"):
raise HTTPException(400, detail="File must be an image")
# Check file size (limit to 2MB)
content = await file.read()
if len(content) > 2 * 1024 * 1024: # 2MB
raise HTTPException(400, detail="File too large (max 2MB)")
# Reset file position after reading for size check
await file.seek(0)
# Process the file...
return {"filename": file.filename, "content_type": file.content_type}
Combining Files with Form Data
In many cases, you'll want to receive files along with other form data:
from fastapi import Form
@app.post("/upload-with-description/")
async def upload_with_description(
file: UploadFile = File(...),
description: str = Form(...),
category: str = Form(None)
):
return {
"filename": file.filename,
"description": description,
"category": category
}
Practical Examples
Example 1: Image Upload Service
Let's build a simple image upload service with size validation, thumbnail generation, and metadata extraction:
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import FileResponse
from PIL import Image, ExifTags
import io
import uuid
from pathlib import Path
import shutil
app = FastAPI()
@app.post("/upload-image/")
async def upload_image_with_processing(image: UploadFile = File(...)):
# Validate image
if not image.content_type.startswith("image/"):
raise HTTPException(400, detail="Not an image")
# Read image content
content = await image.read()
try:
# Process with Pillow
img = Image.open(io.BytesIO(content))
# Extract EXIF data (if available)
exif_data = {}
if hasattr(img, '_getexif') and img._getexif():
for tag, value in img._getexif().items():
if tag in ExifTags.TAGS:
exif_data[ExifTags.TAGS[tag]] = str(value)
# Generate a thumbnail
img.thumbnail((200, 200))
# Save with unique filename
filename = f"{uuid.uuid4()}.{image.filename.split('.')[-1]}"
upload_dir = Path("uploads")
upload_dir.mkdir(exist_ok=True)
thumbnail_path = upload_dir / f"thumb_{filename}"
img.save(thumbnail_path)
# Save original image
original_path = upload_dir / filename
with original_path.open("wb") as f:
f.write(content)
return {
"filename": filename,
"original_size": len(content),
"thumbnail_path": str(thumbnail_path),
"width": img.width,
"height": img.height,
"exif": exif_data
}
except Exception as e:
raise HTTPException(500, detail=f"Image processing error: {str(e)}")
Example 2: Document Upload with Progress Tracking
For larger files, like documents, you might want to implement progress tracking. Here's a simplified version:
from fastapi import FastAPI, File, UploadFile, WebSocket
import asyncio
import json
app = FastAPI()
# Simulate active uploads with their progress
active_uploads = {}
@app.post("/upload-document/")
async def upload_document(document: UploadFile = File(...)):
# Generate upload ID
upload_id = str(uuid.uuid4())
# Read in chunks to simulate progress
total_size = 0
chunk_size = 1024 * 1024 # 1MB chunks
active_uploads[upload_id] = {"progress": 0, "filename": document.filename}
while True:
chunk = await document.read(chunk_size)
if not chunk:
break
# Process chunk...
total_size += len(chunk)
# Update progress (0-100%)
if document.size:
progress = min(100, int(total_size / document.size * 100))
active_uploads[upload_id]["progress"] = progress
# Cleanup
del active_uploads[upload_id]
return {"upload_id": upload_id, "filename": document.filename, "size": total_size}
@app.websocket("/ws/upload-progress/{upload_id}")
async def upload_progress(websocket: WebSocket, upload_id: str):
await websocket.accept()
try:
while upload_id in active_uploads:
await websocket.send_json(active_uploads[upload_id])
await asyncio.sleep(0.5)
# Send 100% as final message
await websocket.send_json({"progress": 100, "filename": active_uploads.get(upload_id, {}).get("filename")})
except:
pass
finally:
await websocket.close()
Best Practices for File Handling
-
Always validate files: Check file types, sizes, and contents to prevent security issues.
-
Use async methods: FastAPI's async support helps handle multiple file uploads efficiently.
-
Consider storage options: Choose appropriate storage (local filesystem, object storage, etc.) based on your application needs.
-
Handle large files properly: For large files, use streaming and chunk processing to avoid memory issues.
-
Implement security measures: Scan for malware, validate file extensions, and use secure storage paths.
-
Set upload limits: Configure your server to limit file sizes to prevent denial-of-service attacks.
Common Challenges and Solutions
Challenge: File Size Limits
FastAPI doesn't impose file size limits directly, but your server might. To handle large files:
# In your application startup
app = FastAPI()
# Add middleware for handling larger files if needed
from starlette.middleware.base import BaseHTTPMiddleware
class LargeFileMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
if request.method == "POST" and "multipart/form-data" in request.headers.get("content-type", ""):
# Allow large uploads for specific endpoints
if request.url.path in ["/upload-large-file/"]:
# Processing for large files
pass
response = await call_next(request)
return response
app.add_middleware(LargeFileMiddleware)
Challenge: File Type Detection
Don't rely solely on file extensions or content-type headers for security:
import magic # python-magic package
@app.post("/secure-upload/")
async def secure_upload(file: UploadFile = File(...)):
content = await file.read()
# Use python-magic for MIME type detection
mime = magic.Magic(mime=True)
detected_type = mime.from_buffer(content)
# Check if claimed content-type matches actual content
if detected_type != file.content_type:
raise HTTPException(400, detail=f"File appears to be {detected_type}, not {file.content_type}")
# Continue processing...
return {"filename": file.filename, "detected_type": detected_type}
Summary
In this guide, we've covered how to handle file uploads in FastAPI applications:
- Basic file uploads with
UploadFile
andFile
- Single and multiple file handling
- File validation and security considerations
- Saving files to the filesystem
- Practical examples like image processing and document uploads
FastAPI's file handling capabilities are powerful and flexible, making it an excellent choice for applications that need to process user uploads. By following the best practices outlined in this guide, you can build secure and efficient file upload features for your web applications.
Additional Resources
- FastAPI Official Documentation on File Uploads
- Python-Multipart Documentation
- Pillow (PIL Fork) Documentation for image processing
- Python Magic for better file type detection
Exercises
-
Create a FastAPI endpoint that accepts multiple image uploads, validates them, and creates thumbnails for each one.
-
Implement a document converter that accepts PDF uploads and extracts text content.
-
Build a simple avatar upload system that crops images to a square and resizes them to standard dimensions.
-
Create a secure file sharing system with expiring download links for uploaded files.
-
Implement a file upload progress tracker using WebSockets to display real-time progress to users.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)