diff --git a/docs/en/docs/tutorial/request-files.md b/docs/en/docs/tutorial/request-files.md index 3d6e9c18a0..a4cc1ef973 100644 --- a/docs/en/docs/tutorial/request-files.md +++ b/docs/en/docs/tutorial/request-files.md @@ -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_py39.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 } Use `File`, `bytes`, and `UploadFile` to declare files to be uploaded in the request, sent as form data. diff --git a/docs_src/request_files/tutorial004.py b/docs_src/request_files/tutorial004.py new file mode 100644 index 0000000000..7dfcf2875c --- /dev/null +++ b/docs_src/request_files/tutorial004.py @@ -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} diff --git a/docs_src/request_files/tutorial004_an_py310.py b/docs_src/request_files/tutorial004_an_py310.py new file mode 100644 index 0000000000..fa1fa61b31 --- /dev/null +++ b/docs_src/request_files/tutorial004_an_py310.py @@ -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} diff --git a/docs_src/request_files/tutorial004_an_py39.py b/docs_src/request_files/tutorial004_an_py39.py new file mode 100644 index 0000000000..fa1fa61b31 --- /dev/null +++ b/docs_src/request_files/tutorial004_an_py39.py @@ -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} diff --git a/docs_src/request_files/tutorial005.py b/docs_src/request_files/tutorial005.py new file mode 100644 index 0000000000..345e301711 --- /dev/null +++ b/docs_src/request_files/tutorial005.py @@ -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} diff --git a/docs_src/request_files/tutorial005_an_py310.py b/docs_src/request_files/tutorial005_an_py310.py new file mode 100644 index 0000000000..cc2c96b601 --- /dev/null +++ b/docs_src/request_files/tutorial005_an_py310.py @@ -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} diff --git a/docs_src/request_files/tutorial005_an_py39.py b/docs_src/request_files/tutorial005_an_py39.py new file mode 100644 index 0000000000..cc2c96b601 --- /dev/null +++ b/docs_src/request_files/tutorial005_an_py39.py @@ -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} diff --git a/tests/test_tutorial/test_request_files/test_tutorial004.py b/tests/test_tutorial/test_request_files/test_tutorial004.py new file mode 100644 index 0000000000..adca95defe --- /dev/null +++ b/tests/test_tutorial/test_request_files/test_tutorial004.py @@ -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"] diff --git a/tests/test_tutorial/test_request_files/test_tutorial005.py b/tests/test_tutorial/test_request_files/test_tutorial005.py new file mode 100644 index 0000000000..8232fa8ea4 --- /dev/null +++ b/tests/test_tutorial/test_request_files/test_tutorial005.py @@ -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"]