0
# File Handling
1
2
FastAPI provides comprehensive file upload and handling capabilities through the UploadFile class and File parameter function. This enables secure file uploads, async file operations, and integration with form data for building file processing applications.
3
4
## Capabilities
5
6
### UploadFile Class
7
8
Async file upload handler providing access to uploaded files with metadata and streaming capabilities.
9
10
```python { .api }
11
class UploadFile:
12
"""
13
File upload handler with async methods for file operations.
14
15
Provides access to uploaded file content, metadata, and streaming
16
operations while maintaining memory efficiency for large files.
17
18
Attributes (read-only):
19
- filename: Original filename from client (can be None)
20
- size: File size in bytes (available after reading)
21
- headers: HTTP headers from multipart upload
22
- content_type: MIME type of the uploaded file
23
"""
24
25
filename: Optional[str]
26
size: Optional[int]
27
headers: Headers
28
content_type: Optional[str]
29
file: BinaryIO # Underlying file-like object
30
31
async def read(self, size: int = -1) -> bytes:
32
"""
33
Read file content as bytes.
34
35
Parameters:
36
- size: Maximum number of bytes to read (-1 for entire file)
37
38
Returns:
39
File content as bytes
40
41
Note: File position advances after reading
42
"""
43
44
async def write(self, data: Union[str, bytes]) -> None:
45
"""
46
Write data to file.
47
48
Parameters:
49
- data: String or bytes to write to file
50
51
Note: Useful for processing or modifying uploaded files
52
"""
53
54
async def seek(self, offset: int) -> None:
55
"""
56
Move file pointer to specific position.
57
58
Parameters:
59
- offset: Position to seek to (0 = beginning)
60
61
Note: Allows re-reading file content or random access
62
"""
63
64
async def close(self) -> None:
65
"""
66
Close the file and release resources.
67
68
Note: Files are automatically closed when request completes
69
"""
70
```
71
72
### File Parameter Function
73
74
Function for declaring file upload parameters in endpoint functions.
75
76
```python { .api }
77
def File(
78
default: Any = Undefined,
79
*,
80
media_type: str = "multipart/form-data",
81
alias: Optional[str] = None,
82
title: Optional[str] = None,
83
description: Optional[str] = None,
84
gt: Optional[float] = None,
85
ge: Optional[float] = None,
86
lt: Optional[float] = None,
87
le: Optional[float] = None,
88
min_length: Optional[int] = None,
89
max_length: Optional[int] = None,
90
regex: Optional[str] = None,
91
example: Any = Undefined,
92
examples: Optional[Dict[str, Any]] = None,
93
deprecated: Optional[bool] = None,
94
include_in_schema: bool = True,
95
json_schema_extra: Union[Dict[str, Any], Callable[[Dict[str, Any]], None], None] = None,
96
**extra: Any
97
) -> Any:
98
"""
99
Declare file upload parameter with validation constraints.
100
101
Parameters:
102
- default: Default value (use ... for required files)
103
- media_type: Expected media type (defaults to multipart/form-data)
104
- alias: Alternative parameter name in OpenAPI schema
105
- title: Parameter title for OpenAPI documentation
106
- description: Parameter description for OpenAPI documentation
107
- gt, ge, lt, le: Numeric validation for file size
108
- min_length, max_length: File size validation
109
- regex: Pattern for filename validation
110
- example: Example value for OpenAPI documentation
111
- examples: Dictionary of examples for OpenAPI documentation
112
- deprecated: Mark parameter as deprecated
113
- include_in_schema: Include in OpenAPI schema
114
- json_schema_extra: Additional JSON schema properties
115
116
Returns:
117
- bytes: When type hint is bytes
118
- UploadFile: When type hint is UploadFile
119
- List[UploadFile]: When type hint is List[UploadFile] for multiple files
120
"""
121
```
122
123
## File Upload Patterns
124
125
### Single File Upload
126
127
```python
128
from fastapi import FastAPI, File, UploadFile
129
from typing import Optional
130
131
app = FastAPI()
132
133
@app.post("/upload-file/")
134
async def upload_file(file: UploadFile = File(...)):
135
"""Upload single file as UploadFile object."""
136
return {
137
"filename": file.filename,
138
"content_type": file.content_type,
139
"size": len(await file.read())
140
}
141
142
@app.post("/upload-bytes/")
143
async def upload_bytes(file: bytes = File(...)):
144
"""Upload single file as bytes."""
145
return {"file_size": len(file)}
146
```
147
148
### Multiple File Upload
149
150
```python
151
from fastapi import FastAPI, File, UploadFile
152
from typing import List
153
154
app = FastAPI()
155
156
@app.post("/upload-multiple/")
157
async def upload_multiple_files(files: List[UploadFile] = File(...)):
158
"""Upload multiple files."""
159
results = []
160
161
for file in files:
162
content = await file.read()
163
results.append({
164
"filename": file.filename,
165
"content_type": file.content_type,
166
"size": len(content)
167
})
168
169
return {"files": results, "total_files": len(files)}
170
171
@app.post("/upload-optional-multiple/")
172
async def upload_optional_multiple(files: Optional[List[UploadFile]] = File(None)):
173
"""Upload multiple files (optional)."""
174
if not files:
175
return {"message": "No files uploaded"}
176
177
return {"files_uploaded": len(files)}
178
```
179
180
## Usage Examples
181
182
### Basic File Upload with Validation
183
184
```python
185
from fastapi import FastAPI, File, UploadFile, HTTPException
186
import os
187
from typing import List
188
189
app = FastAPI()
190
191
# Configuration
192
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB
193
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".pdf", ".txt"}
194
UPLOAD_DIR = "uploads"
195
196
# Ensure upload directory exists
197
os.makedirs(UPLOAD_DIR, exist_ok=True)
198
199
def validate_file(file: UploadFile) -> None:
200
"""Validate uploaded file."""
201
# Check file extension
202
if file.filename:
203
file_ext = os.path.splitext(file.filename)[1].lower()
204
if file_ext not in ALLOWED_EXTENSIONS:
205
raise HTTPException(
206
status_code=400,
207
detail=f"File type {file_ext} not allowed. Allowed types: {ALLOWED_EXTENSIONS}"
208
)
209
210
# Check content type
211
allowed_content_types = {
212
"image/jpeg", "image/png", "image/gif",
213
"application/pdf", "text/plain"
214
}
215
216
if file.content_type not in allowed_content_types:
217
raise HTTPException(
218
status_code=400,
219
detail=f"Content type {file.content_type} not allowed"
220
)
221
222
@app.post("/upload/")
223
async def upload_file(file: UploadFile = File(...)):
224
# Validate file
225
validate_file(file)
226
227
# Read file content
228
content = await file.read()
229
230
# Check file size
231
if len(content) > MAX_FILE_SIZE:
232
raise HTTPException(
233
status_code=413,
234
detail=f"File size {len(content)} exceeds maximum {MAX_FILE_SIZE} bytes"
235
)
236
237
# Save file
238
file_path = os.path.join(UPLOAD_DIR, file.filename)
239
240
with open(file_path, "wb") as f:
241
f.write(content)
242
243
return {
244
"filename": file.filename,
245
"content_type": file.content_type,
246
"size": len(content),
247
"saved_path": file_path
248
}
249
```
250
251
### File Upload with Form Data
252
253
```python
254
from fastapi import FastAPI, File, Form, UploadFile
255
from typing import Optional
256
257
app = FastAPI()
258
259
@app.post("/upload-with-metadata/")
260
async def upload_with_metadata(
261
title: str = Form(...),
262
description: Optional[str] = Form(None),
263
tags: str = Form(...),
264
file: UploadFile = File(...)
265
):
266
"""Upload file with additional form data."""
267
content = await file.read()
268
269
# Process metadata
270
tag_list = [tag.strip() for tag in tags.split(",")]
271
272
# Save file and metadata
273
file_record = {
274
"filename": file.filename,
275
"title": title,
276
"description": description,
277
"tags": tag_list,
278
"content_type": file.content_type,
279
"size": len(content),
280
"upload_date": "2023-01-01T00:00:00Z"
281
}
282
283
# Save to database (your logic here)
284
file_id = save_file_record(file_record, content)
285
286
return {"file_id": file_id, "metadata": file_record}
287
288
@app.post("/upload-multiple-with-metadata/")
289
async def upload_multiple_with_metadata(
290
category: str = Form(...),
291
files: List[UploadFile] = File(...)
292
):
293
"""Upload multiple files with shared metadata."""
294
results = []
295
296
for file in files:
297
content = await file.read()
298
299
file_record = {
300
"filename": file.filename,
301
"category": category,
302
"content_type": file.content_type,
303
"size": len(content)
304
}
305
306
results.append(file_record)
307
308
return {"category": category, "files": results, "total": len(files)}
309
```
310
311
### Stream Processing Large Files
312
313
```python
314
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
315
import hashlib
316
import aiofiles
317
import os
318
319
app = FastAPI()
320
321
async def process_large_file_stream(file: UploadFile, output_path: str):
322
"""Process large file in chunks to avoid memory issues."""
323
hash_md5 = hashlib.md5()
324
total_size = 0
325
326
# Create output file
327
async with aiofiles.open(output_path, "wb") as output_file:
328
# Process file in chunks
329
while True:
330
chunk = await file.read(8192) # 8KB chunks
331
if not chunk:
332
break
333
334
# Update hash and size
335
hash_md5.update(chunk)
336
total_size += len(chunk)
337
338
# Write processed chunk (could apply transformations here)
339
await output_file.write(chunk)
340
341
return {
342
"total_size": total_size,
343
"md5_hash": hash_md5.hexdigest()
344
}
345
346
@app.post("/upload-large/")
347
async def upload_large_file(
348
file: UploadFile = File(...),
349
background_tasks: BackgroundTasks = None
350
):
351
"""Handle large file upload with streaming."""
352
output_path = os.path.join("uploads", f"large_{file.filename}")
353
354
# Process file in streaming fashion
355
result = await process_large_file_stream(file, output_path)
356
357
# Add background task for further processing if needed
358
if background_tasks:
359
background_tasks.add_task(post_process_file, output_path)
360
361
return {
362
"filename": file.filename,
363
"saved_path": output_path,
364
"processing_result": result
365
}
366
367
def post_process_file(file_path: str):
368
"""Background task for additional file processing."""
369
print(f"Post-processing file: {file_path}")
370
# Additional processing logic (virus scan, format conversion, etc.)
371
```
372
373
### Image Upload with Processing
374
375
```python
376
from fastapi import FastAPI, File, UploadFile, HTTPException
377
from PIL import Image
378
import io
379
import os
380
381
app = FastAPI()
382
383
def validate_image(file: UploadFile) -> None:
384
"""Validate that uploaded file is a valid image."""
385
if not file.content_type.startswith("image/"):
386
raise HTTPException(
387
status_code=400,
388
detail="File must be an image"
389
)
390
391
def resize_image(image_data: bytes, max_width: int = 800, max_height: int = 600) -> bytes:
392
"""Resize image while maintaining aspect ratio."""
393
image = Image.open(io.BytesIO(image_data))
394
395
# Calculate new dimensions
396
image.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
397
398
# Save resized image
399
output = io.BytesIO()
400
image.save(output, format=image.format)
401
return output.getvalue()
402
403
@app.post("/upload-image/")
404
async def upload_image(
405
file: UploadFile = File(...),
406
resize: bool = False,
407
max_width: int = 800,
408
max_height: int = 600
409
):
410
"""Upload and optionally resize image."""
411
validate_image(file)
412
413
# Read image data
414
image_data = await file.read()
415
416
# Get image info
417
image = Image.open(io.BytesIO(image_data))
418
original_size = image.size
419
420
# Resize if requested
421
if resize:
422
image_data = resize_image(image_data, max_width, max_height)
423
resized_image = Image.open(io.BytesIO(image_data))
424
new_size = resized_image.size
425
else:
426
new_size = original_size
427
428
# Save processed image
429
filename = f"processed_{file.filename}"
430
file_path = os.path.join("uploads", filename)
431
432
with open(file_path, "wb") as f:
433
f.write(image_data)
434
435
return {
436
"filename": filename,
437
"original_size": {"width": original_size[0], "height": original_size[1]},
438
"final_size": {"width": new_size[0], "height": new_size[1]},
439
"resized": resize,
440
"file_size": len(image_data),
441
"saved_path": file_path
442
}
443
```
444
445
### CSV File Processing
446
447
```python
448
from fastapi import FastAPI, File, UploadFile, HTTPException
449
import csv
450
import io
451
from typing import List, Dict
452
453
app = FastAPI()
454
455
async def process_csv(file: UploadFile) -> List[Dict]:
456
"""Process uploaded CSV file."""
457
if file.content_type != "text/csv":
458
raise HTTPException(
459
status_code=400,
460
detail="File must be a CSV file"
461
)
462
463
# Read file content
464
content = await file.read()
465
466
# Decode and parse CSV
467
csv_data = content.decode("utf-8")
468
csv_reader = csv.DictReader(io.StringIO(csv_data))
469
470
records = []
471
for row_num, row in enumerate(csv_reader, start=2): # Start at 2 (header is row 1)
472
# Validate required fields
473
if not row.get("name"):
474
raise HTTPException(
475
status_code=400,
476
detail=f"Missing 'name' field in row {row_num}"
477
)
478
479
# Process row data
480
processed_row = {
481
"name": row["name"].strip(),
482
"email": row.get("email", "").strip(),
483
"age": int(row["age"]) if row.get("age") and row["age"].isdigit() else None,
484
"row_number": row_num
485
}
486
487
records.append(processed_row)
488
489
return records
490
491
@app.post("/upload-csv/")
492
async def upload_csv(file: UploadFile = File(...)):
493
"""Upload and process CSV file."""
494
try:
495
records = await process_csv(file)
496
497
# Save processed data (your logic here)
498
saved_count = save_csv_records(records)
499
500
return {
501
"filename": file.filename,
502
"total_records": len(records),
503
"saved_records": saved_count,
504
"sample_records": records[:3] # Show first 3 records
505
}
506
507
except Exception as e:
508
raise HTTPException(status_code=400, detail=f"CSV processing error: {str(e)}")
509
```
510
511
### File Download and Response
512
513
```python
514
from fastapi import FastAPI, HTTPException
515
from fastapi.responses import FileResponse, StreamingResponse
516
import os
517
import mimetypes
518
519
app = FastAPI()
520
521
@app.get("/download/{filename}")
522
async def download_file(filename: str):
523
"""Download file by filename."""
524
file_path = os.path.join("uploads", filename)
525
526
if not os.path.exists(file_path):
527
raise HTTPException(status_code=404, detail="File not found")
528
529
# Detect content type
530
content_type, _ = mimetypes.guess_type(file_path)
531
532
return FileResponse(
533
path=file_path,
534
filename=filename,
535
media_type=content_type
536
)
537
538
@app.get("/download-stream/{filename}")
539
async def download_file_stream(filename: str):
540
"""Download large file as stream."""
541
file_path = os.path.join("uploads", filename)
542
543
if not os.path.exists(file_path):
544
raise HTTPException(status_code=404, detail="File not found")
545
546
def iter_file():
547
with open(file_path, "rb") as file:
548
while True:
549
chunk = file.read(8192) # 8KB chunks
550
if not chunk:
551
break
552
yield chunk
553
554
content_type, _ = mimetypes.guess_type(file_path)
555
556
return StreamingResponse(
557
iter_file(),
558
media_type=content_type,
559
headers={"Content-Disposition": f"attachment; filename={filename}"}
560
)
561
```
562
563
### File Upload with Progress Tracking
564
565
```python
566
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
567
import aiofiles
568
import os
569
from typing import Dict
570
import uuid
571
572
app = FastAPI()
573
574
# Store upload progress
575
upload_progress: Dict[str, Dict] = {}
576
577
async def save_file_with_progress(
578
file: UploadFile,
579
file_path: str,
580
upload_id: str
581
):
582
"""Save file with progress tracking."""
583
total_size = 0
584
chunk_size = 8192
585
586
# Initialize progress
587
upload_progress[upload_id] = {
588
"filename": file.filename,
589
"total_bytes": 0,
590
"bytes_written": 0,
591
"status": "uploading",
592
"percentage": 0
593
}
594
595
async with aiofiles.open(file_path, "wb") as output_file:
596
while True:
597
chunk = await file.read(chunk_size)
598
if not chunk:
599
break
600
601
await output_file.write(chunk)
602
total_size += len(chunk)
603
604
# Update progress
605
upload_progress[upload_id]["bytes_written"] = total_size
606
upload_progress[upload_id]["percentage"] = min(100, (total_size / (total_size + 1)) * 100)
607
608
# Mark as complete
609
upload_progress[upload_id]["status"] = "completed"
610
upload_progress[upload_id]["total_bytes"] = total_size
611
upload_progress[upload_id]["percentage"] = 100
612
613
@app.post("/upload-with-progress/")
614
async def upload_with_progress(
615
file: UploadFile = File(...),
616
background_tasks: BackgroundTasks = None
617
):
618
"""Upload file with progress tracking."""
619
upload_id = str(uuid.uuid4())
620
file_path = os.path.join("uploads", f"{upload_id}_{file.filename}")
621
622
# Start upload in background
623
background_tasks.add_task(
624
save_file_with_progress,
625
file,
626
file_path,
627
upload_id
628
)
629
630
return {
631
"upload_id": upload_id,
632
"filename": file.filename,
633
"status": "upload_started"
634
}
635
636
@app.get("/upload-progress/{upload_id}")
637
async def get_upload_progress(upload_id: str):
638
"""Get upload progress by ID."""
639
if upload_id not in upload_progress:
640
raise HTTPException(status_code=404, detail="Upload ID not found")
641
642
return upload_progress[upload_id]
643
```
644
645
### Secure File Upload
646
647
```python
648
from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
649
import hashlib
650
import os
651
import magic # python-magic library for MIME type detection
652
from typing import Set
653
654
app = FastAPI()
655
656
# Security configuration
657
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB
658
ALLOWED_MIME_TYPES: Set[str] = {
659
"image/jpeg", "image/png", "image/gif",
660
"application/pdf", "text/plain"
661
}
662
663
BLOCKED_EXTENSIONS: Set[str] = {
664
".exe", ".bat", ".cmd", ".com", ".pif", ".scr", ".vbs", ".js"
665
}
666
667
def scan_file_security(file_content: bytes, filename: str) -> None:
668
"""Perform security checks on uploaded file."""
669
# Check file extension
670
file_ext = os.path.splitext(filename)[1].lower()
671
if file_ext in BLOCKED_EXTENSIONS:
672
raise HTTPException(
673
status_code=400,
674
detail=f"File extension {file_ext} is not allowed for security reasons"
675
)
676
677
# Detect actual MIME type from content
678
mime_type = magic.from_buffer(file_content, mime=True)
679
680
if mime_type not in ALLOWED_MIME_TYPES:
681
raise HTTPException(
682
status_code=400,
683
detail=f"File type {mime_type} is not allowed"
684
)
685
686
# Check for embedded executables or suspicious patterns
687
suspicious_patterns = [
688
b"MZ", # PE executable header
689
b"\x7fELF", # ELF executable header
690
b"<script", # JavaScript in uploads
691
b"<?php" # PHP code in uploads
692
]
693
694
for pattern in suspicious_patterns:
695
if pattern in file_content:
696
raise HTTPException(
697
status_code=400,
698
detail="File contains suspicious content"
699
)
700
701
def generate_secure_filename(original_filename: str) -> str:
702
"""Generate secure filename to prevent directory traversal."""
703
# Remove directory separators and dangerous characters
704
safe_chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-_"
705
filename = "".join(c for c in original_filename if c in safe_chars)
706
707
# Ensure filename is not empty and has reasonable length
708
if not filename or len(filename) > 255:
709
filename = f"upload_{hashlib.md5(original_filename.encode()).hexdigest()[:8]}"
710
711
return filename
712
713
@app.post("/secure-upload/")
714
async def secure_upload(
715
file: UploadFile = File(...),
716
current_user = Depends(get_current_user) # Require authentication
717
):
718
"""Secure file upload with comprehensive validation."""
719
# Read file content
720
content = await file.read()
721
722
# Check file size
723
if len(content) > MAX_FILE_SIZE:
724
raise HTTPException(
725
status_code=413,
726
detail=f"File size exceeds maximum allowed size of {MAX_FILE_SIZE} bytes"
727
)
728
729
# Perform security scans
730
scan_file_security(content, file.filename)
731
732
# Generate secure filename
733
secure_filename = generate_secure_filename(file.filename)
734
735
# Calculate file hash for integrity
736
file_hash = hashlib.sha256(content).hexdigest()
737
738
# Save file with secure path
739
user_dir = os.path.join("uploads", f"user_{current_user['id']}")
740
os.makedirs(user_dir, exist_ok=True)
741
742
file_path = os.path.join(user_dir, secure_filename)
743
744
with open(file_path, "wb") as f:
745
f.write(content)
746
747
# Log upload for audit
748
log_file_upload(current_user["id"], secure_filename, len(content), file_hash)
749
750
return {
751
"filename": secure_filename,
752
"original_filename": file.filename,
753
"size": len(content),
754
"hash": file_hash,
755
"content_type": file.content_type,
756
"saved_path": file_path
757
}
758
```