mirror of https://github.com/tiangolo/fastapi.git
Merge c340795243 into 31bbb38074
This commit is contained in:
commit
434e9c85cd
|
|
@ -171,6 +171,60 @@ And the same way as before, you can use `File()` to set additional parameters, e
|
||||||
|
|
||||||
{* ../../docs_src/request_files/tutorial003_an_py310.py hl[11,18:20] *}
|
{* ../../docs_src/request_files/tutorial003_an_py310.py hl[11,18:20] *}
|
||||||
|
|
||||||
|
## File Upload with Validation { #file-upload-with-validation }
|
||||||
|
|
||||||
|
In production applications, you often need to validate uploaded files to ensure they meet certain requirements.
|
||||||
|
|
||||||
|
Here's an example that validates:
|
||||||
|
|
||||||
|
* **File type**: Only allows specific image formats (JPEG, PNG, GIF, WebP)
|
||||||
|
* **File size**: Maximum 5MB per file
|
||||||
|
* **Number of files**: Maximum 10 files per request
|
||||||
|
|
||||||
|
{* ../../docs_src/request_files/tutorial004_an_py39.py hl[9:13,22:24,26:30,33:50] *}
|
||||||
|
|
||||||
|
/// tip
|
||||||
|
|
||||||
|
Notice how we:
|
||||||
|
|
||||||
|
1. Check the `content_type` attribute to validate the file type
|
||||||
|
2. Read the file contents with `await file.read()` to check the size
|
||||||
|
3. Use `await file.seek(0)` to reset the file pointer for potential further processing
|
||||||
|
4. Raise `HTTPException` with appropriate status codes and clear error messages
|
||||||
|
|
||||||
|
///
|
||||||
|
|
||||||
|
/// warning
|
||||||
|
|
||||||
|
Be careful when reading file contents to check the size. For very large files, this could consume significant memory. Consider using streaming validation for production applications with very large file uploads.
|
||||||
|
|
||||||
|
///
|
||||||
|
|
||||||
|
## Saving Uploaded Files { #saving-uploaded-files }
|
||||||
|
|
||||||
|
When you need to save uploaded files to disk, you can use Python's built-in file operations along with `shutil` for efficient copying.
|
||||||
|
|
||||||
|
Here's a practical example of uploading and saving product images:
|
||||||
|
|
||||||
|
{* ../../docs_src/request_files/tutorial005_an_py39.py hl[1:4,10:12,16:20,53:59,61:63] *}
|
||||||
|
|
||||||
|
/// tip
|
||||||
|
|
||||||
|
**Security best practices:**
|
||||||
|
|
||||||
|
* Generate unique filenames using `uuid4()` to avoid conflicts and potential security issues
|
||||||
|
* Validate file extensions to prevent malicious uploads
|
||||||
|
* Store files outside your application's source code directory
|
||||||
|
* Consider using a dedicated storage service (like AWS S3, Google Cloud Storage) for production applications
|
||||||
|
|
||||||
|
///
|
||||||
|
|
||||||
|
/// note | Technical Details
|
||||||
|
|
||||||
|
The `shutil.copyfileobj()` function efficiently copies the file content from the `UploadFile.file` object to the destination file. This is more memory-efficient than reading the entire file into memory first.
|
||||||
|
|
||||||
|
///
|
||||||
|
|
||||||
## Recap { #recap }
|
## Recap { #recap }
|
||||||
|
|
||||||
Use `File`, `bytes`, and `UploadFile` to declare files to be uploaded in the request, sent as form data.
|
Use `File`, `bytes`, and `UploadFile` to declare files to be uploaded in the request, sent as form data.
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
from typing import List
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-images/")
|
||||||
|
async def upload_images(
|
||||||
|
files: List[UploadFile] = File(description="Multiple images"),
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
results = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
|
||||||
|
results.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"content_type": file.content_type,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(results), "files": results}
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
from typing import Annotated
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-images/")
|
||||||
|
async def upload_images(
|
||||||
|
files: Annotated[list[UploadFile], File(description="Multiple images")],
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
results = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
|
||||||
|
results.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"content_type": file.content_type,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(results), "files": results}
|
||||||
|
|
@ -0,0 +1,43 @@
|
||||||
|
from typing import Annotated
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-images/")
|
||||||
|
async def upload_images(
|
||||||
|
files: Annotated[list[UploadFile], File(description="Multiple images")],
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
results = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
|
||||||
|
results.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"content_type": file.content_type,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(results), "files": results}
|
||||||
|
|
@ -0,0 +1,55 @@
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import List
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
UPLOAD_DIR = Path("uploaded_files")
|
||||||
|
UPLOAD_DIR.mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-product-images/")
|
||||||
|
async def upload_product_images(
|
||||||
|
files: List[UploadFile] = File(description="Product images"),
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
saved = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
file_ext = Path(file.filename).suffix
|
||||||
|
unique_name = f"{uuid4()}{file_ext}"
|
||||||
|
file_path = UPLOAD_DIR / unique_name
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
with file_path.open("wb") as buffer:
|
||||||
|
shutil.copyfileobj(file.file, buffer)
|
||||||
|
|
||||||
|
saved.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"saved_as": unique_name,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(saved), "files": saved}
|
||||||
|
|
@ -0,0 +1,55 @@
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Annotated
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
UPLOAD_DIR = Path("uploaded_files")
|
||||||
|
UPLOAD_DIR.mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-product-images/")
|
||||||
|
async def upload_product_images(
|
||||||
|
files: Annotated[list[UploadFile], File(description="Product images")],
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
saved = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
file_ext = Path(file.filename).suffix
|
||||||
|
unique_name = f"{uuid4()}{file_ext}"
|
||||||
|
file_path = UPLOAD_DIR / unique_name
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
with file_path.open("wb") as buffer:
|
||||||
|
shutil.copyfileobj(file.file, buffer)
|
||||||
|
|
||||||
|
saved.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"saved_as": unique_name,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(saved), "files": saved}
|
||||||
|
|
@ -0,0 +1,55 @@
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Annotated
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
from fastapi import FastAPI, File, HTTPException, UploadFile
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
|
||||||
|
UPLOAD_DIR = Path("uploaded_files")
|
||||||
|
UPLOAD_DIR.mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/upload-product-images/")
|
||||||
|
async def upload_product_images(
|
||||||
|
files: Annotated[list[UploadFile], File(description="Product images")],
|
||||||
|
):
|
||||||
|
allowed_types = ["image/jpeg", "image/png", "image/webp"]
|
||||||
|
max_size = 5 * 1024 * 1024 # 5MB
|
||||||
|
|
||||||
|
if len(files) > 10:
|
||||||
|
raise HTTPException(status_code=400, detail="Too many files")
|
||||||
|
|
||||||
|
saved = []
|
||||||
|
|
||||||
|
for file in files:
|
||||||
|
if file.content_type not in allowed_types:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail=f"Invalid file type: {file.content_type}",
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = await file.read()
|
||||||
|
if len(contents) > max_size:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400, detail=f"File too large: {file.filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
file_ext = Path(file.filename).suffix
|
||||||
|
unique_name = f"{uuid4()}{file_ext}"
|
||||||
|
file_path = UPLOAD_DIR / unique_name
|
||||||
|
|
||||||
|
await file.seek(0)
|
||||||
|
with file_path.open("wb") as buffer:
|
||||||
|
shutil.copyfileobj(file.file, buffer)
|
||||||
|
|
||||||
|
saved.append(
|
||||||
|
{
|
||||||
|
"filename": file.filename,
|
||||||
|
"saved_as": unique_name,
|
||||||
|
"size": len(contents),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return {"uploaded": len(saved), "files": saved}
|
||||||
|
|
@ -0,0 +1,87 @@
|
||||||
|
import importlib
|
||||||
|
import io
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from ...utils import needs_py39, needs_py310
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(
|
||||||
|
name="app",
|
||||||
|
params=[
|
||||||
|
"tutorial004",
|
||||||
|
pytest.param("tutorial004_an_py39", marks=needs_py39),
|
||||||
|
pytest.param("tutorial004_an_py310", marks=needs_py310),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def get_app(request: pytest.FixtureRequest):
|
||||||
|
mod = importlib.import_module(f"docs_src.request_files.{request.param}")
|
||||||
|
|
||||||
|
return mod.app
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(name="client")
|
||||||
|
def get_client(app: FastAPI):
|
||||||
|
client = TestClient(app)
|
||||||
|
return client
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_images_valid(client: TestClient):
|
||||||
|
# Create fake image files
|
||||||
|
file1 = ("test1.jpg", io.BytesIO(b"fake image content"), "image/jpeg")
|
||||||
|
file2 = ("test2.png", io.BytesIO(b"another fake image"), "image/png")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-images/",
|
||||||
|
files=[
|
||||||
|
("files", file1),
|
||||||
|
("files", file2),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
assert response.status_code == 200, response.text
|
||||||
|
data = response.json()
|
||||||
|
assert data["uploaded"] == 2
|
||||||
|
assert len(data["files"]) == 2
|
||||||
|
assert data["files"][0]["filename"] == "test1.jpg"
|
||||||
|
assert data["files"][0]["content_type"] == "image/jpeg"
|
||||||
|
assert data["files"][1]["filename"] == "test2.png"
|
||||||
|
assert data["files"][1]["content_type"] == "image/png"
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_images_invalid_type(client: TestClient):
|
||||||
|
# Upload a non-image file
|
||||||
|
file1 = ("test.txt", io.BytesIO(b"text content"), "text/plain")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-images/",
|
||||||
|
files=[("files", file1)],
|
||||||
|
)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "Invalid file type" in response.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_images_too_large(client: TestClient):
|
||||||
|
# Create a file larger than 5MB
|
||||||
|
large_content = b"x" * (6 * 1024 * 1024) # 6MB
|
||||||
|
file1 = ("large.jpg", io.BytesIO(large_content), "image/jpeg")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-images/",
|
||||||
|
files=[("files", file1)],
|
||||||
|
)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "too large" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_images_too_many_files(client: TestClient):
|
||||||
|
# Try to upload 11 files (max is 10)
|
||||||
|
files = [
|
||||||
|
("files", (f"test{i}.jpg", io.BytesIO(b"content"), "image/jpeg"))
|
||||||
|
for i in range(11)
|
||||||
|
]
|
||||||
|
|
||||||
|
response = client.post("/upload-images/", files=files)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "Too many files" in response.json()["detail"]
|
||||||
|
|
@ -0,0 +1,88 @@
|
||||||
|
import importlib
|
||||||
|
import io
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from ...utils import needs_py39, needs_py310
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(
|
||||||
|
name="app",
|
||||||
|
params=[
|
||||||
|
"tutorial005",
|
||||||
|
pytest.param("tutorial005_an_py39", marks=needs_py39),
|
||||||
|
pytest.param("tutorial005_an_py310", marks=needs_py310),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def get_app(request: pytest.FixtureRequest):
|
||||||
|
mod = importlib.import_module(f"docs_src.request_files.{request.param}")
|
||||||
|
|
||||||
|
return mod.app
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(name="client")
|
||||||
|
def get_client(app: FastAPI):
|
||||||
|
client = TestClient(app)
|
||||||
|
return client
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_product_images_valid(tmp_path, client: TestClient):
|
||||||
|
# Create fake image files
|
||||||
|
file1 = ("product1.jpg", io.BytesIO(b"fake image content"), "image/jpeg")
|
||||||
|
file2 = ("product2.png", io.BytesIO(b"another fake image"), "image/png")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-product-images/",
|
||||||
|
files=[
|
||||||
|
("files", file1),
|
||||||
|
("files", file2),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
assert response.status_code == 200, response.text
|
||||||
|
data = response.json()
|
||||||
|
assert data["uploaded"] == 2
|
||||||
|
assert len(data["files"]) == 2
|
||||||
|
assert data["files"][0]["filename"] == "product1.jpg"
|
||||||
|
assert data["files"][1]["filename"] == "product2.png"
|
||||||
|
# Check that saved_as is a UUID-like string
|
||||||
|
assert len(data["files"][0]["saved_as"]) > 30
|
||||||
|
assert len(data["files"][1]["saved_as"]) > 30
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_product_images_invalid_type(client: TestClient):
|
||||||
|
# Upload a non-image file
|
||||||
|
file1 = ("test.txt", io.BytesIO(b"text content"), "text/plain")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-product-images/",
|
||||||
|
files=[("files", file1)],
|
||||||
|
)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "Invalid file type" in response.json()["detail"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_product_images_too_large(client: TestClient):
|
||||||
|
# Create a file larger than 5MB
|
||||||
|
large_content = b"x" * (6 * 1024 * 1024) # 6MB
|
||||||
|
file1 = ("large.jpg", io.BytesIO(large_content), "image/jpeg")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/upload-product-images/",
|
||||||
|
files=[("files", file1)],
|
||||||
|
)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "too large" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_upload_product_images_too_many_files(client: TestClient):
|
||||||
|
# Try to upload 11 files (max is 10)
|
||||||
|
files = [
|
||||||
|
("files", (f"product{i}.jpg", io.BytesIO(b"content"), "image/jpeg"))
|
||||||
|
for i in range(11)
|
||||||
|
]
|
||||||
|
|
||||||
|
response = client.post("/upload-product-images/", files=files)
|
||||||
|
assert response.status_code == 400, response.text
|
||||||
|
assert "Too many files" in response.json()["detail"]
|
||||||
Loading…
Reference in New Issue