File Uploads and Downloads — Large Files and Bulk Imports
Uploading files is like receiving a package—small packages are signed for immediately (memory), while large shipments require unloading in batches (streamed writes); downloading is like shipping a package—single packages (FileResponse), and bulk shipments travel on a conveyor belt (StreamingResponse).
1. What You'll Learn
UploadFileandFile()parameters: Stream-based upload vs. in-memory upload- Large file processing: block-by-block reading,
shutil.copyfileobjstreaming write - CSV/Excel Parsing:
pandas+ Pydantic Model Joint Validation - File response:
FileResponse/StreamingResponsedownload endpoints - Alice Scenario: Bob uploads a CSV file containing millions of rows of price data → Asynchronous parsing by Celery in the background → Notification upon completion of the import
2. Alice's True Story
(1) Pain Point: Out-of-memory error when importing a CSV file with one million rows
Bob uploads a CSV file containing a million rows of price data to PriceTracker every month. The previous implementation read the entire file into memory before parsing it, and a 1GB CSV file caused the Python process to crash due to an OOM (Out of Memory) error. To make matters worse, the CSV contained invalid data (negative prices, invalid currencies), and after the import failed halfway through, the database was left in an inconsistent state.
(2) Solution Using Stream-Based Upload and Asynchronous Processing
FastAPI's UploadFile saves files as temporary files by default (which do not consume memory). Combined with Celery for asynchronous processing of millions of lines of data, Pydantic for line-by-line validation to skip invalid data, and transactional batch inserts to ensure consistency.
(3) Revenue
The 1GB CSV import has gone from crashing due to an Out-of-Memory (OOM) error to running stably, with memory usage dropping from 2GB to 50MB. A Celery Worker processes one million rows in about 10 minutes, and Bob can view the progress in real time via the task status endpoint.
3. UploadFile Basics
(1) UploadFile vs. bytes
(1) ▶ Example: Simple File Upload
PYTHON
from fastapi import FastAPI, UploadFile, File, HTTPException
app = FastAPI()
@app.post("/upload/single")
async def upload_single(file: UploadFile = File(...)):
# UploadFile: file stored as temp file, NOT in memory
content = await file.read()
return {
"filename": file.filename,
"size": len(content),
"content_type": file.content_type,
}
Output:
TEXT
# Function defined successfully
| Method | Memory Usage | Suitable File Size | API |
|---|---|---|---|
bytes |
Load all into memory | < 2MB | file: bytes = File() |
UploadFile |
Temporary files (streaming) | Unlimited | file: UploadFile = File() |
(2) ▶ Example: Uploading Multiple Files
PYTHON
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
results = []
for file in files:
content = await file.read()
results.append({
"filename": file.filename,
"size": len(content),
})
return {"uploaded": len(results), "files": results}
Output:
TEXT
# Function defined successfully
4. Streaming Processing of Large Files
(1) Block-by-Block Reading and Stream-Based Writing
(1) ▶ Example: Streaming Large Files
PYTHON
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload/large")
async def upload_large_file(file: UploadFile = File(...)):
# Stream file to disk - never loads entire file into memory
dest = UPLOAD_DIR / file.filename
with open(dest, "wb") as buffer:
# Copy in chunks (default 64KB buffer)
shutil.copyfileobj(file.file, buffer)
file_size = dest.stat().st_size
return {
"filename": file.filename,
"size_bytes": file_size,
"saved_to": str(dest),
}
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Customizing the chunk size
PYTHON
CHUNK_SIZE = 1024 * 1024 # 1MB chunks
@app.post("/upload/chunked")
async def upload_chunked(file: UploadFile = File(...)):
dest = UPLOAD_DIR / file.filename
bytes_written = 0
with open(dest, "wb") as buffer:
while chunk := await file.read(CHUNK_SIZE):
buffer.write(chunk)
bytes_written += len(chunk)
return {"filename": file.filename, "bytes_written": bytes_written}
Output:
TEXT
# Function defined successfully
5. CSV/Excel Bulk Import
(1) Large File Upload + Asynchronous Processing Workflow
sequenceDiagram
participant Bob as Bob Frontend
participant API as FastAPI
participant Temp as Temp File
participant Celery as Celery Worker
participant DB as PostgreSQL
Bob->>API: POST /import/csv (UploadFile)
API->>Temp: Save to temp file
API-->>Bob: 202 Accepted + task_id
API->>Celery: Trigger parse task
Celery->>Temp: Read CSV in chunks
Celery->>Celery: Validate each row (Pydantic)
Celery->>DB: Batch insert valid rows
Celery-->>Celery: Report progress
Bob->>API: GET /tasks/{task_id}
API-->>Bob: Progress: 75%
Celery-->>Celery: Task complete
Bob->>API: GET /tasks/{task_id}
API-->>Bob: Status: SUCCESS, imported: 950000
(1) ▶ Example: CSV Upload Endpoint + Celery Task Trigger
PYTHON
import csv
import io
from fastapi import FastAPI, UploadFile, File, Depends
from app.tasks import import_prices_from_csv
app = FastAPI()
UPLOAD_DIR = Path("uploads")
@app.post("/api/v1/import/csv")
async def import_csv(
file: UploadFile = File(..., description="CSV file with price data"),
user=Depends(require_subscription("pro")),
):
if not file.filename.endswith(".csv"):
raise HTTPException(status_code=400, detail="Only CSV files accepted")
# Save uploaded file
dest = UPLOAD_DIR / f"{uuid4()}.csv"
with open(dest, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Trigger Celery task for async processing
task = import_prices_from_csv.delay(str(dest), user_id=user.id)
return {
"task_id": task.id,
"filename": file.filename,
"status": "processing",
"message": "File uploaded. Check task status for progress.",
}
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Celery task that parses a CSV file line by line
PYTHON
# app/tasks/import_tasks.py
from app.core.celery_app import celery_app
from pydantic import BaseModel, Field, field_validator
import csv
class PriceRow(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0)
currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")
source: str = Field(max_length=100, default="csv_import")
@field_validator("price")
@classmethod
def round_price(cls, v: float) -> float:
return round(v, 2)
@celery_app.task(bind=True)
def import_prices_from_csv(self, file_path: str, user_id: int):
valid_rows = []
invalid_rows = []
total_rows = 0
with open(file_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
total_rows += 1
try:
validated = PriceRow(**row)
valid_rows.append(validated.model_dump())
except Exception as e:
invalid_rows.append({"row": total_rows, "error": str(e)})
# Report progress every 10000 rows
if total_rows % 10000 == 0:
self.update_state(
state="PROGRESS",
meta={"current": total_rows, "valid": len(valid_rows), "invalid": len(invalid_rows)},
)
# Batch insert valid rows
batch_insert_prices(valid_rows, batch_size=5000)
# Clean up temp file
Path(file_path).unlink(missing_ok=True)
return {
"total": total_rows,
"imported": len(valid_rows),
"skipped": len(invalid_rows),
}
Output:
TEXT
# Function defined successfully
(2) File Format Processing Matrix
| Format | Parsing Library | Advantages | Disadvantages |
|---|---|---|---|
| CSV | CSV / pandas | Lightweight, stream-based | No type or encoding issues |
| XLSX | openpyxl / pandas | Type-aware, multiple sheets | High memory usage |
| JSON | json / orjson | Structured, Pydantic-friendly | Large file size |
| Parquet | pyarrow | Columnar storage, high compression ratio | Requires additional libraries |
6. File Downloads
(1) FileResponse and StreamingResponse
(1) ▶ Example: FileResponse—Downloading a File
PYTHON
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pathlib import Path
app = FastAPI()
@app.get("/download/prices/csv")
async def download_prices_csv(
user=Depends(require_subscription("pro")),
):
# Generate CSV file (or use pre-generated)
file_path = Path("exports/prices.csv")
return FileResponse(
path=file_path,
filename="price_data.csv",
media_type="text/csv",
)
Output:
TEXT
# Function defined successfully
(2) ▶ Example: StreamingResponse—Streaming CSV Generation
PYTHON
from fastapi.responses import StreamingResponse
import csv
import io
from app.core.deps import get_db
@app.get("/api/v1/export/prices")
async def export_prices(
category: str | None = None,
db: AsyncSession = Depends(get_db),
user=Depends(require_subscription("pro")),
):
async def generate_csv():
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["product_id", "name", "price", "currency", "recorded_at"])
yield output.getvalue()
output.seek(0)
output.truncate(0)
# Stream rows in batches
offset = 0
batch_size = 5000
while True:
rows = await fetch_price_batch(db, category, offset, batch_size)
if not rows:
break
for row in rows:
writer.writerow([
row.product_id, row.name,
row.price, row.currency, row.recorded_at,
])
yield output.getvalue()
output.seek(0)
output.truncate(0)
offset += batch_size
return StreamingResponse(
generate_csv(),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=prices.csv"},
)
Output:
TEXT
# Function defined successfully
| Response Type | Use Case | Memory Usage |
|---|---|---|
FileResponse |
Existing files | Low (OS-level streaming) |
StreamingResponse |
Dynamic Generation | Very Low (Generated Line by Line) |
❓ FAQ
Q When are UploadFile temporary files deleted?
A They are automatically deleted after the request ends. If you need to retain them, copy them to a persistent directory during the upload process.
Q Is there a limit on file size for uploads?
A FastAPI has no framework-imposed limits, but Uvicorn imposes a default limit on request body size. In a production environment, this is controlled by Nginx's
client_max_body_size.Q What should I do if some rows are invalid during a CSV import?
A The approach depends on your business needs—strict mode (roll back all invalid rows) or lenient mode (skip invalid rows and import valid ones). PriceTracker uses lenient mode and returns the number of imported rows and the number of skipped rows.
Q How should Excel files be handled?
A Use pandas'
read_excel() to parse them, but XLSX files cannot be read in a streaming manner (they must be loaded in their entirety). For files with millions of rows, it is recommended to convert them to CSV format before uploading.Q Does the generator for
StreamingResponse have to be async?A No, it doesn't have to be; a synchronous generator is also acceptable. However, an
async generator does not block the event loop, so it is recommended.Q How can I restrict the types of files that can be uploaded?
A Check
file.content_type (e.g., text/csv) and the file extension. Note that the content_type can be spoofed; use this only for front-end prompts—server-side parsing and validation are still required.📖 Summary
UploadFileStreams data to a temporary file, doesn't use memory, and is suitable for files of any size- Use
shutil.copyfileobj()for large files orread()for chunked streaming to disk - CSV Upload + Celery Asynchronous Parsing: The API immediately returns a
task_id; the background process validates each row and performs batch insertion - Pydantic validates CSV data line by line, skips and logs invalid lines, and uses a lenient mode to ensure partial imports
FileResponseDownload existing files;StreamingResponseDynamically generate and return data in real time
📝 Exercises
- Basic Problem (Difficulty ⭐): Implement a CSV upload endpoint that accepts
UploadFile, reads the content, and returns the number of rows and column names. Hint:file: UploadFile = File(...)+csv.DictReader(io.StringIO(content)) - Advanced Exercise (Difficulty ⭐⭐): Implement a CSV upload → Celery asynchronous parsing workflow: Save the uploaded file to a temporary directory, trigger a Celery task, return a
task_id, and use another endpoint to query the task status and import results. Hint:shutil.copyfileobj+import_prices.delay(file_path) - Challenge (Difficulty: ⭐⭐⭐): Implement a StreamingResponse export endpoint: Stream price data from the database, generate and return CSV data row by row, keep memory usage under 1 MB, and support filtering by product category. Hint:
async def generate()+yield+StreamingResponse(generate(), media_type="text/csv")
---|
