从零开始学习如何使用FastAPI与Pillow构建高性能图像处理API并解决实际开发中遇到的各种挑战的完整指南

威震华夏关云长

1. 引言

FastAPI是一个现代、高性能的Web框架，用于构建API，基于Python 3.6+的类型提示。它具有自动交互式文档、数据验证、依赖注入等强大功能。Pillow是Python图像处理库（PIL）的一个分支，提供了广泛的图像处理功能。结合这两个库，我们可以构建高性能的图像处理API，用于各种应用，如社交媒体、电子商务、内容管理系统等。

本指南将带你从零开始，逐步学习如何使用FastAPI与Pillow构建高性能图像处理API，并解决实际开发中可能遇到的各种挑战。

2. 环境设置

在开始之前，我们需要安装Python和必要的库。

2.1 安装Python

首先，确保你的系统上安装了Python 3.6或更高版本。你可以从Python官网下载并安装Python。

2.2 创建虚拟环境

创建一个虚拟环境来隔离项目依赖：

2. # 创建项目目录
3. mkdir fastapi-image-processing
4. cd fastapi-image-processing
6. # 创建虚拟环境
7. python -m venv venv
9. # 激活虚拟环境
10. # Windows
11. venv\Scripts\activate
12. # macOS/Linux
13. source venv/bin/activate

复制代码

2.3 安装必要的库

安装FastAPI、Pillow和其他必要的库：

2. pip install fastapi "uvicorn[standard]" pillow python-multipart aiofiles

复制代码

这些库的作用如下：

• fastapi: Web框架
• uvicorn: ASGI服务器，用于运行FastAPI应用
• pillow: 图像处理库
• python-multipart: 处理表单数据，包括文件上传
• aiofiles: 异步文件操作

3. FastAPI基础

3.1 创建基本的FastAPI应用

创建一个名为main.py的文件，并添加以下代码：

2. from fastapi import FastAPI
4. app = FastAPI(title="图像处理API", description="使用FastAPI和Pillow构建的高性能图像处理API")
6. @app.get("/")
7. async def root():
8.     return {"message": "欢迎使用图像处理API"}
10. if __name__ == "__main__":
11.     import uvicorn
12.     uvicorn.run(app, host="0.0.0.0", port=8000)

复制代码

3.2 运行FastAPI应用

在终端中运行以下命令启动应用：

2. python main.py

复制代码

或者使用uvicorn直接运行：

2. uvicorn main:app --reload

复制代码

--reload选项使服务器在代码更改时自动重启。

3.3 访问API文档

打开浏览器，访问http://localhost:8000/docs，你将看到自动生成的交互式API文档。

3.4 添加路由

让我们添加一些基本的路由：

2. from fastapi import FastAPI
3. from pydantic import BaseModel
5. app = FastAPI(title="图像处理API", description="使用FastAPI和Pillow构建的高性能图像处理API")
7. class HealthCheck(BaseModel):
8.     status: str
9.     message: str
11. @app.get("/", response_model=HealthCheck)
12. async def root():
13.     return {"status": "ok", "message": "欢迎使用图像处理API"}
15. @app.get("/health", response_model=HealthCheck)
16. async def health_check():
17.     return {"status": "ok", "message": "API运行正常"}

复制代码

4. Pillow基础

4.1 基本图像操作

创建一个名为image_utils.py的文件，并添加以下代码：

2. from PIL import Image, ImageFilter, ImageEnhance
3. import io
4. from typing import Tuple, Optional
6. def open_image(image_bytes: bytes) -> Image.Image:
7.     """从字节流打开图像"""
8.     return Image.open(io.BytesIO(image_bytes))
10. def save_image(image: Image.Image, format: str = "JPEG", quality: int = 85) -> bytes:
11.     """将图像保存为字节流"""
12.     img_byte_arr = io.BytesIO()
13.     image.save(img_byte_arr, format=format, quality=quality)
14.     img_byte_arr.seek(0)
15.     return img_byte_arr.getvalue()
17. def resize_image(image: Image.Image, size: Tuple[int, int], resample: int = Image.LANCZOS) -> Image.Image:
18.     """调整图像大小"""
19.     return image.resize(size, resample)
21. def crop_image(image: Image.Image, box: Tuple[int, int, int, int]) -> Image.Image:
22.     """裁剪图像"""
23.     return image.crop(box)
25. def rotate_image(image: Image.Image, angle: float, expand: bool = True) -> Image.Image:
26.     """旋转图像"""
27.     return image.rotate(angle, expand=expand)
29. def flip_image(image: Image.Image, mode: str = "horizontal") -> Image.Image:
30.     """翻转图像"""
31.     if mode == "horizontal":
32.         return image.transpose(Image.FLIP_LEFT_RIGHT)
33.     elif mode == "vertical":
34.         return image.transpose(Image.FLIP_TOP_BOTTOM)
35.     else:
36.         raise ValueError("模式必须是'horizontal'或'vertical'")
38. def apply_filter(image: Image.Image, filter_type: str) -> Image.Image:
39.     """应用滤镜"""
40.     filters = {
41.         "blur": ImageFilter.BLUR,
42.         "sharpen": ImageFilter.SHARPEN,
43.         "edge_enhance": ImageFilter.EDGE_ENHANCE,
44.         "emboss": ImageFilter.EMBOSS,
45.         "smooth": ImageFilter.SMOOTH,
46.     }
47.     if filter_type not in filters:
48.         raise ValueError(f"不支持的滤镜类型: {filter_type}")
49.     return image.filter(filters[filter_type])
51. def adjust_brightness(image: Image.Image, factor: float) -> Image.Image:
52.     """调整亮度"""
53.     enhancer = ImageEnhance.Brightness(image)
54.     return enhancer.enhance(factor)
56. def adjust_contrast(image: Image.Image, factor: float) -> Image.Image:
57.     """调整对比度"""
58.     enhancer = ImageEnhance.Contrast(image)
59.     return enhancer.enhance(factor)
61. def adjust_saturation(image: Image.Image, factor: float) -> Image.Image:
62.     """调整饱和度"""
63.     enhancer = ImageEnhance.Color(image)
64.     return enhancer.enhance(factor)
66. def convert_format(image: Image.Image, format: str) -> Image.Image:
67.     """转换图像格式"""
68.     if format.upper() == "JPEG" and image.mode in ("RGBA", "LA", "P"):
69.         # JPEG不支持透明度，转换为RGB
70.         background = Image.new("RGB", image.size, (255, 255, 255))
71.         background.paste(image, mask=image.split()[-1] if image.mode == "RGBA" else None)
72.         image = background
73.     elif format.upper() == "PNG" and image.mode not in ("RGBA", "LA", "P"):
74.         # PNG支持透明度，转换为RGBA
75.         image = image.convert("RGBA")
76.     return image

复制代码

4.2 测试图像处理函数

创建一个名为test_image_utils.py的文件，并添加以下代码来测试我们的图像处理函数：

2. from PIL import Image
3. import requests
4. from image_utils import (
5.     open_image, save_image, resize_image, crop_image, rotate_image,
6.     flip_image, apply_filter, adjust_brightness, adjust_contrast,
7.     adjust_saturation, convert_format
8. )
10. # 下载测试图像
11. url = "https://picsum.photos/800/600.jpg"
12. response = requests.get(url)
13. image_bytes = response.content
15. # 打开图像
16. image = open_image(image_bytes)
17. print(f"原始图像尺寸: {image.size}, 模式: {image.mode}")
19. # 调整大小
20. resized_image = resize_image(image, (400, 300))
21. print(f"调整大小后的图像尺寸: {resized_image.size}")
23. # 裁剪图像
24. cropped_image = crop_image(image, (100, 100, 700, 500))
25. print(f"裁剪后的图像尺寸: {cropped_image.size}")
27. # 旋转图像
28. rotated_image = rotate_image(image, 45)
29. print(f"旋转后的图像尺寸: {rotated_image.size}")
31. # 翻转图像
32. flipped_image = flip_image(image, "horizontal")
33. print(f"翻转后的图像尺寸: {flipped_image.size}")
35. # 应用滤镜
36. filtered_image = apply_filter(image, "blur")
37. print(f"应用滤镜后的图像尺寸: {filtered_image.size}")
39. # 调整亮度
40. brightened_image = adjust_brightness(image, 1.5)
41. print(f"调整亮度后的图像尺寸: {brightened_image.size}")
43. # 调整对比度
44. contrasted_image = adjust_contrast(image, 1.5)
45. print(f"调整对比度后的图像尺寸: {contrasted_image.size}")
47. # 调整饱和度
48. saturated_image = adjust_saturation(image, 1.5)
49. print(f"调整饱和度后的图像尺寸: {saturated_image.size}")
51. # 转换格式
52. png_image = convert_format(image, "PNG")
53. print(f"转换格式后的图像模式: {png_image.mode}")
55. # 保存处理后的图像
56. save_image(resized_image, "resized.jpg")
57. save_image(cropped_image, "cropped.jpg")
58. save_image(rotated_image, "rotated.jpg")
59. save_image(flipped_image, "flipped.jpg")
60. save_image(filtered_image, "filtered.jpg")
61. save_image(brightened_image, "brightened.jpg")
62. save_image(contrasted_image, "contrasted.jpg")
63. save_image(saturated_image, "saturated.jpg")
64. save_image(png_image, "converted.png")
66. print("测试完成！")

复制代码

运行测试文件：

2. python test_image_utils.py

复制代码

5. 构建图像处理API

现在，我们将FastAPI和Pillow结合起来，构建一个图像处理API。

5.1 创建图像处理路由

在main.py中添加以下代码：

2. from fastapi import FastAPI, File, UploadFile, HTTPException, Form, Query
3. from fastapi.responses import StreamingResponse
4. from pydantic import BaseModel
5. from typing import Optional, List
6. import io
7. from image_utils import (
8.     open_image, save_image, resize_image, crop_image, rotate_image,
9.     flip_image, apply_filter, adjust_brightness, adjust_contrast,
10.     adjust_saturation, convert_format
11. )
13. app = FastAPI(title="图像处理API", description="使用FastAPI和Pillow构建的高性能图像处理API")
15. class HealthCheck(BaseModel):
16.     status: str
17.     message: str
19. @app.get("/", response_model=HealthCheck)
20. async def root():
21.     return {"status": "ok", "message": "欢迎使用图像处理API"}
23. @app.get("/health", response_model=HealthCheck)
24. async def health_check():
25.     return {"status": "ok", "message": "API运行正常"}
27. @app.post("/process/")
28. async def process_image(
29.     file: UploadFile = File(...),
30.     operation: str = Form(...),
31.     width: Optional[int] = Form(None),
32.     height: Optional[int] = Form(None),
33.     angle: Optional[float] = Form(None),
34.     flip_mode: Optional[str] = Form(None),
35.     filter_type: Optional[str] = Form(None),
36.     brightness: Optional[float] = Form(None),
37.     contrast: Optional[float] = Form(None),
38.     saturation: Optional[float] = Form(None),
39.     output_format: Optional[str] = Form("JPEG"),
40.     quality: Optional[int] = Form(85),
41.     left: Optional[int] = Form(None),
42.     top: Optional[int] = Form(None),
43.     right: Optional[int] = Form(None),
44.     bottom: Optional[int] = Form(None)
45. ):
46.     """处理图像"""
47.     # 检查文件类型
48.     if not file.content_type.startswith("image/"):
49.         raise HTTPException(status_code=400, detail="文件必须是图像")
50.    
51.     try:
52.         # 读取图像
53.         image_bytes = await file.read()
54.         image = open_image(image_bytes)
55.         
56.         # 执行操作
57.         if operation == "resize":
58.             if width is None or height is None:
59.                 raise HTTPException(status_code=400, detail="调整大小需要宽度和高度参数")
60.             image = resize_image(image, (width, height))
61.         elif operation == "crop":
62.             if left is None or top is None or right is None or bottom is None:
63.                 raise HTTPException(status_code=400, detail="裁剪需要left、top、right和bottom参数")
64.             image = crop_image(image, (left, top, right, bottom))
65.         elif operation == "rotate":
66.             if angle is None:
67.                 raise HTTPException(status_code=400, detail="旋转需要角度参数")
68.             image = rotate_image(image, angle)
69.         elif operation == "flip":
70.             if flip_mode is None:
71.                 raise HTTPException(status_code=400, detail="翻转需要flip_mode参数")
72.             image = flip_image(image, flip_mode)
73.         elif operation == "filter":
74.             if filter_type is None:
75.                 raise HTTPException(status_code=400, detail="滤镜需要filter_type参数")
76.             image = apply_filter(image, filter_type)
77.         elif operation == "brightness":
78.             if brightness is None:
79.                 raise HTTPException(status_code=400, detail="亮度调整需要brightness参数")
80.             image = adjust_brightness(image, brightness)
81.         elif operation == "contrast":
82.             if contrast is None:
83.                 raise HTTPException(status_code=400, detail="对比度调整需要contrast参数")
84.             image = adjust_contrast(image, contrast)
85.         elif operation == "saturation":
86.             if saturation is None:
87.                 raise HTTPException(status_code=400, detail="饱和度调整需要saturation参数")
88.             image = adjust_saturation(image, saturation)
89.         elif operation == "convert":
90.             if output_format is None:
91.                 raise HTTPException(status_code=400, detail="格式转换需要output_format参数")
92.             image = convert_format(image, output_format)
93.         else:
94.             raise HTTPException(status_code=400, detail=f"不支持的操作: {operation}")
95.         
96.         # 转换格式
97.         if output_format:
98.             image = convert_format(image, output_format)
99.         
100.         # 保存图像
101.         processed_image_bytes = save_image(image, format=output_format, quality=quality)
102.         
103.         # 返回处理后的图像
104.         return StreamingResponse(
105.             io.BytesIO(processed_image_bytes),
106.             media_type=f"image/{output_format.lower()}",
107.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
108.         )
109.    
110.     except Exception as e:
111.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

5.2 批量处理端点

添加一个批量处理端点，可以一次对多个图像应用相同的操作：

2. @app.post("/batch-process/")
3. async def batch_process_images(
4.     files: List[UploadFile] = File(...),
5.     operation: str = Form(...),
6.     width: Optional[int] = Form(None),
7.     height: Optional[int] = Form(None),
8.     angle: Optional[float] = Form(None),
9.     flip_mode: Optional[str] = Form(None),
10.     filter_type: Optional[str] = Form(None),
11.     brightness: Optional[float] = Form(None),
12.     contrast: Optional[float] = Form(None),
13.     saturation: Optional[float] = Form(None),
14.     output_format: Optional[str] = Form("JPEG"),
15.     quality: Optional[int] = Form(85),
16.     left: Optional[int] = Form(None),
17.     top: Optional[int] = Form(None),
18.     right: Optional[int] = Form(None),
19.     bottom: Optional[int] = Form(None)
20. ):
21.     """批量处理图像"""
22.     results = []
23.    
24.     for file in files:
25.         # 检查文件类型
26.         if not file.content_type.startswith("image/"):
27.             results.append({"filename": file.filename, "error": "文件必须是图像"})
28.             continue
29.         
30.         try:
31.             # 读取图像
32.             image_bytes = await file.read()
33.             image = open_image(image_bytes)
34.             
35.             # 执行操作
36.             if operation == "resize":
37.                 if width is None or height is None:
38.                     results.append({"filename": file.filename, "error": "调整大小需要宽度和高度参数"})
39.                     continue
40.                 image = resize_image(image, (width, height))
41.             elif operation == "crop":
42.                 if left is None or top is None or right is None or bottom is None:
43.                     results.append({"filename": file.filename, "error": "裁剪需要left、top、right和bottom参数"})
44.                     continue
45.                 image = crop_image(image, (left, top, right, bottom))
46.             elif operation == "rotate":
47.                 if angle is None:
48.                     results.append({"filename": file.filename, "error": "旋转需要角度参数"})
49.                     continue
50.                 image = rotate_image(image, angle)
51.             elif operation == "flip":
52.                 if flip_mode is None:
53.                     results.append({"filename": file.filename, "error": "翻转需要flip_mode参数"})
54.                     continue
55.                 image = flip_image(image, flip_mode)
56.             elif operation == "filter":
57.                 if filter_type is None:
58.                     results.append({"filename": file.filename, "error": "滤镜需要filter_type参数"})
59.                     continue
60.                 image = apply_filter(image, filter_type)
61.             elif operation == "brightness":
62.                 if brightness is None:
63.                     results.append({"filename": file.filename, "error": "亮度调整需要brightness参数"})
64.                     continue
65.                 image = adjust_brightness(image, brightness)
66.             elif operation == "contrast":
67.                 if contrast is None:
68.                     results.append({"filename": file.filename, "error": "对比度调整需要contrast参数"})
69.                     continue
70.                 image = adjust_contrast(image, contrast)
71.             elif operation == "saturation":
72.                 if saturation is None:
73.                     results.append({"filename": file.filename, "error": "饱和度调整需要saturation参数"})
74.                     continue
75.                 image = adjust_saturation(image, saturation)
76.             elif operation == "convert":
77.                 if output_format is None:
78.                     results.append({"filename": file.filename, "error": "格式转换需要output_format参数"})
79.                     continue
80.                 image = convert_format(image, output_format)
81.             else:
82.                 results.append({"filename": file.filename, "error": f"不支持的操作: {operation}"})
83.                 continue
84.             
85.             # 转换格式
86.             if output_format:
87.                 image = convert_format(image, output_format)
88.             
89.             # 保存图像
90.             processed_image_bytes = save_image(image, format=output_format, quality=quality)
91.             
92.             # 添加到结果列表
93.             results.append({
94.                 "filename": file.filename,
95.                 "status": "success",
96.                 "size": image.size,
97.                 "format": image.format,
98.                 "mode": image.mode
99.             })
100.             
101.         except Exception as e:
102.             results.append({"filename": file.filename, "error": str(e)})
103.    
104.     return {"results": results}

复制代码

5.3 图像信息端点

添加一个端点，用于获取图像的基本信息：

2. @app.post("/info/")
3. async def get_image_info(file: UploadFile = File(...)):
4.     """获取图像信息"""
5.     # 检查文件类型
6.     if not file.content_type.startswith("image/"):
7.         raise HTTPException(status_code=400, detail="文件必须是图像")
8.    
9.     try:
10.         # 读取图像
11.         image_bytes = await file.read()
12.         image = open_image(image_bytes)
13.         
14.         # 获取图像信息
15.         info = {
16.             "filename": file.filename,
17.             "format": image.format,
18.             "mode": image.mode,
19.             "size": image.size,
20.             "width": image.width,
21.             "height": image.height,
22.             "content_type": file.content_type,
23.         }
24.         
25.         # 如果有EXIF数据，添加到信息中
26.         if hasattr(image, '_getexif') and image._getexif() is not None:
27.             info["exif"] = image._getexif()
28.         
29.         return info
30.    
31.     except Exception as e:
32.         raise HTTPException(status_code=500, detail=f"获取图像信息时出错: {str(e)}")

复制代码

6. 高级功能

6.1 添加水印

在image_utils.py中添加添加水印的功能：

2. def add_watermark(
3.     image: Image.Image,
4.     watermark_text: str,
5.     position: str = "bottom-right",
6.     opacity: float = 0.5,
7.     font_size: int = 36,
8.     color: Tuple[int, int, int] = (255, 255, 255)
9. ) -> Image.Image:
10.     """添加水印"""
11.     from PIL import ImageDraw, ImageFont
12.    
13.     # 创建一个透明层
14.     watermark = Image.new("RGBA", image.size, (0, 0, 0, 0))
15.     draw = ImageDraw.Draw(watermark)
16.    
17.     # 尝试加载字体，如果失败则使用默认字体
18.     try:
19.         font = ImageFont.truetype("arial.ttf", font_size)
20.     except:
21.         font = ImageFont.load_default()
22.    
23.     # 获取文本大小
24.     text_width, text_height = draw.textsize(watermark_text, font=font)
25.    
26.     # 计算位置
27.     margin = 20
28.     if position == "top-left":
29.         position = (margin, margin)
30.     elif position == "top-right":
31.         position = (image.width - text_width - margin, margin)
32.     elif position == "bottom-left":
33.         position = (margin, image.height - text_height - margin)
34.     elif position == "bottom-right":
35.         position = (image.width - text_width - margin, image.height - text_height - margin)
36.     elif position == "center":
37.         position = ((image.width - text_width) // 2, (image.height - text_height) // 2)
38.     else:
39.         raise ValueError("位置必须是'top-left'、'top-right'、'bottom-left'、'bottom-right'或'center'")
40.    
41.     # 绘制文本
42.     draw.text(position, watermark_text, font=font, fill=(*color, int(255 * opacity)))
43.    
44.     # 合并图像
45.     if image.mode != "RGBA":
46.         image = image.convert("RGBA")
47.    
48.     return Image.alpha_composite(image, watermark)

复制代码

6.2 添加水印端点

在main.py中添加添加水印的端点：

2. @app.post("/watermark/")
3. async def add_watermark_to_image(
4.     file: UploadFile = File(...),
5.     text: str = Form(...),
6.     position: str = Form("bottom-right"),
7.     opacity: float = Form(0.5),
8.     font_size: int = Form(36),
9.     color_r: int = Form(255),
10.     color_g: int = Form(255),
11.     color_b: int = Form(255),
12.     output_format: Optional[str] = Form("JPEG"),
13.     quality: Optional[int] = Form(85)
14. ):
15.     """添加水印"""
16.     # 检查文件类型
17.     if not file.content_type.startswith("image/"):
18.         raise HTTPException(status_code=400, detail="文件必须是图像")
19.    
20.     try:
21.         # 读取图像
22.         image_bytes = await file.read()
23.         image = open_image(image_bytes)
24.         
25.         # 添加水印
26.         watermarked_image = add_watermark(
27.             image,
28.             text,
29.             position=position,
30.             opacity=opacity,
31.             font_size=font_size,
32.             color=(color_r, color_g, color_b)
33.         )
34.         
35.         # 转换格式
36.         if output_format:
37.             watermarked_image = convert_format(watermarked_image, output_format)
38.         
39.         # 保存图像
40.         processed_image_bytes = save_image(watermarked_image, format=output_format, quality=quality)
41.         
42.         # 返回处理后的图像
43.         return StreamingResponse(
44.             io.BytesIO(processed_image_bytes),
45.             media_type=f"image/{output_format.lower()}",
46.             headers={"Content-Disposition": f"attachment; filename=watermarked.{output_format.lower()}"}
47.         )
48.    
49.     except Exception as e:
50.         raise HTTPException(status_code=500, detail=f"添加水印时出错: {str(e)}")

复制代码

6.3 图像合成

在image_utils.py中添加图像合成的功能：

2. def composite_images(
3.     background: Image.Image,
4.     foreground: Image.Image,
5.     position: Tuple[int, int] = (0, 0),
6.     opacity: float = 1.0
7. ) -> Image.Image:
8.     """合成图像"""
9.     # 确保前景图像有透明通道
10.     if foreground.mode != "RGBA":
11.         foreground = foreground.convert("RGBA")
12.    
13.     # 调整透明度
14.     if opacity < 1.0:
15.         foreground = foreground.copy()
16.         alpha = foreground.split()[3]
17.         alpha = Image.eval(alpha, lambda a: int(a * opacity))
18.         foreground.putalpha(alpha)
19.    
20.     # 确保背景图像足够大
21.     bg_width, bg_height = background.size
22.     fg_width, fg_height = foreground.size
23.    
24.     if position[0] + fg_width > bg_width or position[1] + fg_height > bg_height:
25.         raise ValueError("前景图像超出背景图像边界")
26.    
27.     # 确保背景图像有透明通道
28.     if background.mode != "RGBA":
29.         background = background.convert("RGBA")
30.    
31.     # 创建一个临时图像
32.     temp = Image.new("RGBA", background.size, (0, 0, 0, 0))
33.     temp.paste(foreground, position)
34.    
35.     # 合成图像
36.     return Image.alpha_composite(background, temp)

复制代码

6.4 图像合成端点

在main.py中添加图像合成的端点：

2. @app.post("/composite/")
3. async def composite_images_endpoint(
4.     background: UploadFile = File(...),
5.     foreground: UploadFile = File(...),
6.     position_x: int = Form(0),
7.     position_y: int = Form(0),
8.     opacity: float = Form(1.0),
9.     output_format: Optional[str] = Form("PNG"),
10.     quality: Optional[int] = Form(85)
11. ):
12.     """合成图像"""
13.     # 检查文件类型
14.     if not background.content_type.startswith("image/") or not foreground.content_type.startswith("image/"):
15.         raise HTTPException(status_code=400, detail "文件必须是图像")
16.    
17.     try:
18.         # 读取图像
19.         background_bytes = await background.read()
20.         background_image = open_image(background_bytes)
21.         
22.         foreground_bytes = await foreground.read()
23.         foreground_image = open_image(foreground_bytes)
24.         
25.         # 合成图像
26.         composite_image = composite_images(
27.             background_image,
28.             foreground_image,
29.             position=(position_x, position_y),
30.             opacity=opacity
31.         )
32.         
33.         # 转换格式
34.         if output_format:
35.             composite_image = convert_format(composite_image, output_format)
36.         
37.         # 保存图像
38.         processed_image_bytes = save_image(composite_image, format=output_format, quality=quality)
39.         
40.         # 返回处理后的图像
41.         return StreamingResponse(
42.             io.BytesIO(processed_image_bytes),
43.             media_type=f"image/{output_format.lower()}",
44.             headers={"Content-Disposition": f"attachment; filename=composite.{output_format.lower()}"}
45.         )
46.    
47.     except Exception as e:
48.         raise HTTPException(status_code=500, detail=f"合成图像时出错: {str(e)}")

复制代码

7. 性能优化

7.1 异步处理

FastAPI支持异步请求处理，我们可以利用这一点来提高性能。让我们修改我们的图像处理函数，使其支持异步操作。

创建一个名为async_image_utils.py的文件：

2. from PIL import Image, ImageFilter, ImageEnhance
3. import io
4. from typing import Tuple, Optional
5. import asyncio
6. import concurrent.futures
8. # 创建一个线程池执行器
9. executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
11. async def open_image_async(image_bytes: bytes) -> Image.Image:
12.     """从字节流异步打开图像"""
13.     loop = asyncio.get_event_loop()
14.     return await loop.run_in_executor(executor, lambda: Image.open(io.BytesIO(image_bytes)))
16. async def save_image_async(image: Image.Image, format: str = "JPEG", quality: int = 85) -> bytes:
17.     """异步将图像保存为字节流"""
18.     loop = asyncio.get_event_loop()
19.     return await loop.run_in_executor(executor, lambda: _save_image_sync(image, format, quality))
21. def _save_image_sync(image: Image.Image, format: str, quality: int) -> bytes:
22.     """同步保存图像"""
23.     img_byte_arr = io.BytesIO()
24.     image.save(img_byte_arr, format=format, quality=quality)
25.     img_byte_arr.seek(0)
26.     return img_byte_arr.getvalue()
28. async def resize_image_async(image: Image.Image, size: Tuple[int, int], resample: int = Image.LANCZOS) -> Image.Image:
29.     """异步调整图像大小"""
30.     loop = asyncio.get_event_loop()
31.     return await loop.run_in_executor(executor, lambda: image.resize(size, resample))
33. async def crop_image_async(image: Image.Image, box: Tuple[int, int, int, int]) -> Image.Image:
34.     """异步裁剪图像"""
35.     loop = asyncio.get_event_loop()
36.     return await loop.run_in_executor(executor, lambda: image.crop(box))
38. async def rotate_image_async(image: Image.Image, angle: float, expand: bool = True) -> Image.Image:
39.     """异步旋转图像"""
40.     loop = asyncio.get_event_loop()
41.     return await loop.run_in_executor(executor, lambda: image.rotate(angle, expand=expand))
43. async def flip_image_async(image: Image.Image, mode: str = "horizontal") -> Image.Image:
44.     """异步翻转图像"""
45.     loop = asyncio.get_event_loop()
46.     if mode == "horizontal":
47.         return await loop.run_in_executor(executor, lambda: image.transpose(Image.FLIP_LEFT_RIGHT))
48.     elif mode == "vertical":
49.         return await loop.run_in_executor(executor, lambda: image.transpose(Image.FLIP_TOP_BOTTOM))
50.     else:
51.         raise ValueError("模式必须是'horizontal'或'vertical'")
53. async def apply_filter_async(image: Image.Image, filter_type: str) -> Image.Image:
54.     """异步应用滤镜"""
55.     filters = {
56.         "blur": ImageFilter.BLUR,
57.         "sharpen": ImageFilter.SHARPEN,
58.         "edge_enhance": ImageFilter.EDGE_ENHANCE,
59.         "emboss": ImageFilter.EMBOSS,
60.         "smooth": ImageFilter.SMOOTH,
61.     }
62.     if filter_type not in filters:
63.         raise ValueError(f"不支持的滤镜类型: {filter_type}")
64.     loop = asyncio.get_event_loop()
65.     return await loop.run_in_executor(executor, lambda: image.filter(filters[filter_type]))
67. async def adjust_brightness_async(image: Image.Image, factor: float) -> Image.Image:
68.     """异步调整亮度"""
69.     loop = asyncio.get_event_loop()
70.     return await loop.run_in_executor(executor, lambda: ImageEnhance.Brightness(image).enhance(factor))
72. async def adjust_contrast_async(image: Image.Image, factor: float) -> Image.Image:
73.     """异步调整对比度"""
74.     loop = asyncio.get_event_loop()
75.     return await loop.run_in_executor(executor, lambda: ImageEnhance.Contrast(image).enhance(factor))
77. async def adjust_saturation_async(image: Image.Image, factor: float) -> Image.Image:
78.     """异步调整饱和度"""
79.     loop = asyncio.get_event_loop()
80.     return await loop.run_in_executor(executor, lambda: ImageEnhance.Color(image).enhance(factor))
82. async def convert_format_async(image: Image.Image, format: str) -> Image.Image:
83.     """异步转换图像格式"""
84.     loop = asyncio.get_event_loop()
85.     return await loop.run_in_executor(executor, lambda: _convert_format_sync(image, format))
87. def _convert_format_sync(image: Image.Image, format: str) -> Image.Image:
88.     """同步转换图像格式"""
89.     if format.upper() == "JPEG" and image.mode in ("RGBA", "LA", "P"):
90.         # JPEG不支持透明度，转换为RGB
91.         background = Image.new("RGB", image.size, (255, 255, 255))
92.         background.paste(image, mask=image.split()[-1] if image.mode == "RGBA" else None)
93.         image = background
94.     elif format.upper() == "PNG" and image.mode not in ("RGBA", "LA", "P"):
95.         # PNG支持透明度，转换为RGBA
96.         image = image.convert("RGBA")
97.     return image
99. async def add_watermark_async(
100.     image: Image.Image,
101.     watermark_text: str,
102.     position: str = "bottom-right",
103.     opacity: float = 0.5,
104.     font_size: int = 36,
105.     color: Tuple[int, int, int] = (255, 255, 255)
106. ) -> Image.Image:
107.     """异步添加水印"""
108.     loop = asyncio.get_event_loop()
109.     return await loop.run_in_executor(executor, lambda: _add_watermark_sync(image, watermark_text, position, opacity, font_size, color))
111. def _add_watermark_sync(
112.     image: Image.Image,
113.     watermark_text: str,
114.     position: str,
115.     opacity: float,
116.     font_size: int,
117.     color: Tuple[int, int, int]
118. ) -> Image.Image:
119.     """同步添加水印"""
120.     from PIL import ImageDraw, ImageFont
121.    
122.     # 创建一个透明层
123.     watermark = Image.new("RGBA", image.size, (0, 0, 0, 0))
124.     draw = ImageDraw.Draw(watermark)
125.    
126.     # 尝试加载字体，如果失败则使用默认字体
127.     try:
128.         font = ImageFont.truetype("arial.ttf", font_size)
129.     except:
130.         font = ImageFont.load_default()
131.    
132.     # 获取文本大小
133.     text_width, text_height = draw.textsize(watermark_text, font=font)
134.    
135.     # 计算位置
136.     margin = 20
137.     if position == "top-left":
138.         pos = (margin, margin)
139.     elif position == "top-right":
140.         pos = (image.width - text_width - margin, margin)
141.     elif position == "bottom-left":
142.         pos = (margin, image.height - text_height - margin)
143.     elif position == "bottom-right":
144.         pos = (image.width - text_width - margin, image.height - text_height - margin)
145.     elif position == "center":
146.         pos = ((image.width - text_width) // 2, (image.height - text_height) // 2)
147.     else:
148.         raise ValueError("位置必须是'top-left'、'top-right'、'bottom-left'、'bottom-right'或'center'")
149.    
150.     # 绘制文本
151.     draw.text(pos, watermark_text, font=font, fill=(*color, int(255 * opacity)))
152.    
153.     # 合并图像
154.     if image.mode != "RGBA":
155.         image = image.convert("RGBA")
156.    
157.     return Image.alpha_composite(image, watermark)
159. async def composite_images_async(
160.     background: Image.Image,
161.     foreground: Image.Image,
162.     position: Tuple[int, int] = (0, 0),
163.     opacity: float = 1.0
164. ) -> Image.Image:
165.     """异步合成图像"""
166.     loop = asyncio.get_event_loop()
167.     return await loop.run_in_executor(executor, lambda: _composite_images_sync(background, foreground, position, opacity))
169. def _composite_images_sync(
170.     background: Image.Image,
171.     foreground: Image.Image,
172.     position: Tuple[int, int],
173.     opacity: float
174. ) -> Image.Image:
175.     """同步合成图像"""
176.     # 确保前景图像有透明通道
177.     if foreground.mode != "RGBA":
178.         foreground = foreground.convert("RGBA")
179.    
180.     # 调整透明度
181.     if opacity < 1.0:
182.         foreground = foreground.copy()
183.         alpha = foreground.split()[3]
184.         alpha = Image.eval(alpha, lambda a: int(a * opacity))
185.         foreground.putalpha(alpha)
186.    
187.     # 确保背景图像足够大
188.     bg_width, bg_height = background.size
189.     fg_width, fg_height = foreground.size
190.    
191.     if position[0] + fg_width > bg_width or position[1] + fg_height > bg_height:
192.         raise ValueError("前景图像超出背景图像边界")
193.    
194.     # 确保背景图像有透明通道
195.     if background.mode != "RGBA":
196.         background = background.convert("RGBA")
197.    
198.     # 创建一个临时图像
199.     temp = Image.new("RGBA", background.size, (0, 0, 0, 0))
200.     temp.paste(foreground, position)
201.    
202.     # 合成图像
203.     return Image.alpha_composite(background, temp)

复制代码

7.2 创建异步处理端点

在main.py中添加使用异步图像处理函数的端点：

2. from fastapi import FastAPI, File, UploadFile, HTTPException, Form, BackgroundTasks
3. from fastapi.responses import StreamingResponse
4. from pydantic import BaseModel
5. from typing import Optional, List, Dict
6. import io
7. import uuid
8. import os
9. from async_image_utils import (
10.     open_image_async, save_image_async, resize_image_async, crop_image_async, rotate_image_async,
11.     flip_image_async, apply_filter_async, adjust_brightness_async, adjust_contrast_async,
12.     adjust_saturation_async, convert_format_async, add_watermark_async, composite_images_async
13. )
15. app = FastAPI(title="图像处理API", description="使用FastAPI和Pillow构建的高性能图像处理API")
17. # 创建临时目录
18. TEMP_DIR = "temp_images"
19. os.makedirs(TEMP_DIR, exist_ok=True)
21. # 存储处理任务的状态
22. processing_tasks: Dict[str, Dict] = {}
24. class HealthCheck(BaseModel):
25.     status: str
26.     message: str
28. class ProcessingStatus(BaseModel):
29.     task_id: str
30.     status: str
31.     message: str
32.     result_url: Optional[str] = None
34. @app.get("/", response_model=HealthCheck)
35. async def root():
36.     return {"status": "ok", "message": "欢迎使用图像处理API"}
38. @app.get("/health", response_model=HealthCheck)
39. async def health_check():
40.     return {"status": "ok", "message": "API运行正常"}
42. @app.post("/process-async/")
43. async def process_image_async(
44.     background_tasks: BackgroundTasks,
45.     file: UploadFile = File(...),
46.     operation: str = Form(...),
47.     width: Optional[int] = Form(None),
48.     height: Optional[int] = Form(None),
49.     angle: Optional[float] = Form(None),
50.     flip_mode: Optional[str] = Form(None),
51.     filter_type: Optional[str] = Form(None),
52.     brightness: Optional[float] = Form(None),
53.     contrast: Optional[float] = Form(None),
54.     saturation: Optional[float] = Form(None),
55.     output_format: Optional[str] = Form("JPEG"),
56.     quality: Optional[int] = Form(85),
57.     left: Optional[int] = Form(None),
58.     top: Optional[int] = Form(None),
59.     right: Optional[int] = Form(None),
60.     bottom: Optional[int] = Form(None)
61. ):
62.     """异步处理图像"""
63.     # 检查文件类型
64.     if not file.content_type.startswith("image/"):
65.         raise HTTPException(status_code=400, detail="文件必须是图像")
66.    
67.     # 创建任务ID
68.     task_id = str(uuid.uuid4())
69.    
70.     # 初始化任务状态
71.     processing_tasks[task_id] = {
72.         "status": "processing",
73.         "message": "图像处理中",
74.         "result_url": None
75.     }
76.    
77.     # 添加后台任务
78.     background_tasks.add_task(
79.         process_image_task,
80.         task_id,
81.         file,
82.         operation,
83.         width,
84.         height,
85.         angle,
86.         flip_mode,
87.         filter_type,
88.         brightness,
89.         contrast,
90.         saturation,
91.         output_format,
92.         quality,
93.         left,
94.         top,
95.         right,
96.         bottom
97.     )
98.    
99.     return {"task_id": task_id, "status": "processing", "message": "图像处理已开始"}
101. async def process_image_task(
102.     task_id: str,
103.     file: UploadFile,
104.     operation: str,
105.     width: Optional[int],
106.     height: Optional[int],
107.     angle: Optional[float],
108.     flip_mode: Optional[str],
109.     filter_type: Optional[str],
110.     brightness: Optional[float],
111.     contrast: Optional[float],
112.     saturation: Optional[float],
113.     output_format: Optional[str],
114.     quality: Optional[int],
115.     left: Optional[int],
116.     top: Optional[int],
117.     right: Optional[int],
118.     bottom: Optional[int]
119. ):
120.     """处理图像任务"""
121.     try:
122.         # 读取图像
123.         image_bytes = await file.read()
124.         image = await open_image_async(image_bytes)
125.         
126.         # 执行操作
127.         if operation == "resize":
128.             if width is None or height is None:
129.                 processing_tasks[task_id] = {
130.                     "status": "error",
131.                     "message": "调整大小需要宽度和高度参数",
132.                     "result_url": None
133.                 }
134.                 return
135.             image = await resize_image_async(image, (width, height))
136.         elif operation == "crop":
137.             if left is None or top is None or right is None or bottom is None:
138.                 processing_tasks[task_id] = {
139.                     "status": "error",
140.                     "message": "裁剪需要left、top、right和bottom参数",
141.                     "result_url": None
142.                 }
143.                 return
144.             image = await crop_image_async(image, (left, top, right, bottom))
145.         elif operation == "rotate":
146.             if angle is None:
147.                 processing_tasks[task_id] = {
148.                     "status": "error",
149.                     "message": "旋转需要角度参数",
150.                     "result_url": None
151.                 }
152.                 return
153.             image = await rotate_image_async(image, angle)
154.         elif operation == "flip":
155.             if flip_mode is None:
156.                 processing_tasks[task_id] = {
157.                     "status": "error",
158.                     "message": "翻转需要flip_mode参数",
159.                     "result_url": None
160.                 }
161.                 return
162.             image = await flip_image_async(image, flip_mode)
163.         elif operation == "filter":
164.             if filter_type is None:
165.                 processing_tasks[task_id] = {
166.                     "status": "error",
167.                     "message": "滤镜需要filter_type参数",
168.                     "result_url": None
169.                 }
170.                 return
171.             image = await apply_filter_async(image, filter_type)
172.         elif operation == "brightness":
173.             if brightness is None:
174.                 processing_tasks[task_id] = {
175.                     "status": "error",
176.                     "message": "亮度调整需要brightness参数",
177.                     "result_url": None
178.                 }
179.                 return
180.             image = await adjust_brightness_async(image, brightness)
181.         elif operation == "contrast":
182.             if contrast is None:
183.                 processing_tasks[task_id] = {
184.                     "status": "error",
185.                     "message": "对比度调整需要contrast参数",
186.                     "result_url": None
187.                 }
188.                 return
189.             image = await adjust_contrast_async(image, contrast)
190.         elif operation == "saturation":
191.             if saturation is None:
192.                 processing_tasks[task_id] = {
193.                     "status": "error",
194.                     "message": "饱和度调整需要saturation参数",
195.                     "result_url": None
196.                 }
197.                 return
198.             image = await adjust_saturation_async(image, saturation)
199.         elif operation == "convert":
200.             if output_format is None:
201.                 processing_tasks[task_id] = {
202.                     "status": "error",
203.                     "message": "格式转换需要output_format参数",
204.                     "result_url": None
205.                 }
206.                 return
207.             image = await convert_format_async(image, output_format)
208.         else:
209.             processing_tasks[task_id] = {
210.                 "status": "error",
211.                 "message": f"不支持的操作: {operation}",
212.                 "result_url": None
213.             }
214.             return
215.         
216.         # 转换格式
217.         if output_format:
218.             image = await convert_format_async(image, output_format)
219.         
220.         # 保存图像
221.         processed_image_bytes = await save_image_async(image, format=output_format, quality=quality)
222.         
223.         # 保存到临时文件
224.         result_filename = f"{task_id}.{output_format.lower()}"
225.         result_path = os.path.join(TEMP_DIR, result_filename)
226.         
227.         with open(result_path, "wb") as f:
228.             f.write(processed_image_bytes)
229.         
230.         # 更新任务状态
231.         processing_tasks[task_id] = {
232.             "status": "completed",
233.             "message": "图像处理完成",
234.             "result_url": f"/results/{result_filename}"
235.         }
236.    
237.     except Exception as e:
238.         processing_tasks[task_id] = {
239.             "status": "error",
240.             "message": f"处理图像时出错: {str(e)}",
241.             "result_url": None
242.         }
244. @app.get("/status/{task_id}", response_model=ProcessingStatus)
245. async def get_processing_status(task_id: str):
246.     """获取处理状态"""
247.     if task_id not in processing_tasks:
248.         raise HTTPException(status_code=404, detail="任务不存在")
249.    
250.     task = processing_tasks[task_id]
251.     return {
252.         "task_id": task_id,
253.         "status": task["status"],
254.         "message": task["message"],
255.         "result_url": task.get("result_url")
256.     }
258. @app.get("/results/{filename}")
259. async def get_result_image(filename: str):
260.     """获取处理后的图像"""
261.     file_path = os.path.join(TEMP_DIR, filename)
262.    
263.     if not os.path.exists(file_path):
264.         raise HTTPException(status_code=404, detail="文件不存在")
265.    
266.     return FileResponse(file_path)

复制代码

7.3 缓存处理结果

添加缓存机制，避免重复处理相同的图像：

2. from fastapi import FastAPI, Request, Response
3. from fastapi.responses import StreamingResponse
4. import hashlib
5. import time
7. # 创建缓存目录
8. CACHE_DIR = "cache"
9. os.makedirs(CACHE_DIR, exist_ok=True)
11. # 缓存处理结果
12. def get_cache_key(image_bytes: bytes, operation: str, params: dict) -> str:
13.     """生成缓存键"""
14.     # 创建包含图像数据和参数的字符串
15.     data_str = f"{image_bytes}{operation}{str(sorted(params.items()))}"
16.     # 计算MD5哈希作为缓存键
17.     return hashlib.md5(data_str.encode()).hexdigest()
19. def get_cached_result(cache_key: str) -> Optional[bytes]:
20.     """获取缓存结果"""
21.     cache_file = os.path.join(CACHE_DIR, f"{cache_key}")
22.     if os.path.exists(cache_file):
23.         with open(cache_file, "rb") as f:
24.             return f.read()
25.     return None
27. def cache_result(cache_key: str, result: bytes) -> None:
28.     """缓存结果"""
29.     cache_file = os.path.join(CACHE_DIR, f"{cache_key}")
30.     with open(cache_file, "wb") as f:
31.         f.write(result)
33. @app.post("/process-cached/")
34. async def process_image_with_cache(
35.     file: UploadFile = File(...),
36.     operation: str = Form(...),
37.     width: Optional[int] = Form(None),
38.     height: Optional[int] = Form(None),
39.     angle: Optional[float] = Form(None),
40.     flip_mode: Optional[str] = Form(None),
41.     filter_type: Optional[str] = Form(None),
42.     brightness: Optional[float] = Form(None),
43.     contrast: Optional[float] = Form(None),
44.     saturation: Optional[float] = Form(None),
45.     output_format: Optional[str] = Form("JPEG"),
46.     quality: Optional[int] = Form(85),
47.     left: Optional[int] = Form(None),
48.     top: Optional[int] = Form(None),
49.     right: Optional[int] = Form(None),
50.     bottom: Optional[int] = Form(None)
51. ):
52.     """带缓存的图像处理"""
53.     # 检查文件类型
54.     if not file.content_type.startswith("image/"):
55.         raise HTTPException(status_code=400, detail="文件必须是图像")
56.    
57.     try:
58.         # 读取图像
59.         image_bytes = await file.read()
60.         
61.         # 准备参数
62.         params = {
63.             "width": width,
64.             "height": height,
65.             "angle": angle,
66.             "flip_mode": flip_mode,
67.             "filter_type": filter_type,
68.             "brightness": brightness,
69.             "contrast": contrast,
70.             "saturation": saturation,
71.             "output_format": output_format,
72.             "quality": quality,
73.             "left": left,
74.             "top": top,
75.             "right": right,
76.             "bottom": bottom
77.         }
78.         
79.         # 生成缓存键
80.         cache_key = get_cache_key(image_bytes, operation, params)
81.         
82.         # 检查缓存
83.         cached_result = get_cached_result(cache_key)
84.         if cached_result:
85.             return StreamingResponse(
86.                 io.BytesIO(cached_result),
87.                 media_type=f"image/{output_format.lower()}",
88.                 headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}", "X-Cache": "HIT"}
89.             )
90.         
91.         # 打开图像
92.         image = await open_image_async(image_bytes)
93.         
94.         # 执行操作
95.         if operation == "resize":
96.             if width is None or height is None:
97.                 raise HTTPException(status_code=400, detail="调整大小需要宽度和高度参数")
98.             image = await resize_image_async(image, (width, height))
99.         elif operation == "crop":
100.             if left is None or top is None or right is None or bottom is None:
101.                 raise HTTPException(status_code=400, detail="裁剪需要left、top、right和bottom参数")
102.             image = await crop_image_async(image, (left, top, right, bottom))
103.         elif operation == "rotate":
104.             if angle is None:
105.                 raise HTTPException(status_code=400, detail="旋转需要角度参数")
106.             image = await rotate_image_async(image, angle)
107.         elif operation == "flip":
108.             if flip_mode is None:
109.                 raise HTTPException(status_code=400, detail="翻转需要flip_mode参数")
110.             image = await flip_image_async(image, flip_mode)
111.         elif operation == "filter":
112.             if filter_type is None:
113.                 raise HTTPException(status_code=400, detail="滤镜需要filter_type参数")
114.             image = await apply_filter_async(image, filter_type)
115.         elif operation == "brightness":
116.             if brightness is None:
117.                 raise HTTPException(status_code=400, detail="亮度调整需要brightness参数")
118.             image = await adjust_brightness_async(image, brightness)
119.         elif operation == "contrast":
120.             if contrast is None:
121.                 raise HTTPException(status_code=400, detail="对比度调整需要contrast参数")
122.             image = await adjust_contrast_async(image, contrast)
123.         elif operation == "saturation":
124.             if saturation is None:
125.                 raise HTTPException(status_code=400, detail="饱和度调整需要saturation参数")
126.             image = await adjust_saturation_async(image, saturation)
127.         elif operation == "convert":
128.             if output_format is None:
129.                 raise HTTPException(status_code=400, detail="格式转换需要output_format参数")
130.             image = await convert_format_async(image, output_format)
131.         else:
132.             raise HTTPException(status_code=400, detail=f"不支持的操作: {operation}")
133.         
134.         # 转换格式
135.         if output_format:
136.             image = await convert_format_async(image, output_format)
137.         
138.         # 保存图像
139.         processed_image_bytes = await save_image_async(image, format=output_format, quality=quality)
140.         
141.         # 缓存结果
142.         cache_result(cache_key, processed_image_bytes)
143.         
144.         # 返回处理后的图像
145.         return StreamingResponse(
146.             io.BytesIO(processed_image_bytes),
147.             media_type=f"image/{output_format.lower()}",
148.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}", "X-Cache": "MISS"}
149.         )
150.    
151.     except Exception as e:
152.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

8. 部署

8.1 使用Docker部署

创建一个Dockerfile：

2. # 使用Python 3.9作为基础镜像
3. FROM python:3.9-slim
5. # 设置工作目录
6. WORKDIR /app
8. # 安装系统依赖
9. RUN apt-get update && apt-get install -y \
10.     gcc \
11.     && rm -rf /var/lib/apt/lists/*
13. # 复制requirements文件
14. COPY requirements.txt .
16. # 安装Python依赖
17. RUN pip install --no-cache-dir -r requirements.txt
19. # 复制应用代码
20. COPY . .
22. # 创建临时目录
23. RUN mkdir -p temp_images cache
25. # 暴露端口
26. EXPOSE 8000
28. # 运行应用
29. CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

复制代码

创建一个requirements.txt文件：

2. fastapi==0.68.0
3. uvicorn[standard]==0.15.0
4. pillow==8.3.1
5. python-multipart==0.0.5
6. aiofiles==0.7.0
7. python-jose[cryptography]==3.3.0
8. passlib[bcrypt]==1.7.4

复制代码

构建和运行Docker容器：

2. # 构建镜像
3. docker build -t fastapi-image-processing .
5. # 运行容器
6. docker run -d -p 8000:8000 -v $(pwd)/temp_images:/app/temp_images -v $(pwd)/cache:/app/cache --name image-api fastapi-image-processing

复制代码

8.2 使用Docker Compose部署

创建一个docker-compose.yml文件：

2. version: '3.8'
4. services:
5.   api:
6.     build: .
7.     ports:
8.       - "8000:8000"
9.     volumes:
10.       - ./temp_images:/app/temp_images
11.       - ./cache:/app/cache
12.     environment:
13.       - ENVIRONMENT=production
14.     restart: unless-stopped
15.     healthcheck:
16.       test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
17.       interval: 30s
18.       timeout: 10s
19.       retries: 3
20.       start_period: 40s
22.   nginx:
23.     image: nginx:alpine
24.     ports:
25.       - "80:80"
26.     volumes:
27.       - ./nginx.conf:/etc/nginx/nginx.conf:ro
28.     depends_on:
29.       - api
30.     restart: unless-stopped

复制代码

创建一个nginx.conf文件：

2. events {
3.     worker_connections 1024;
4. }
6. http {
7.     upstream api {
8.         server api:8000;
9.     }
11.     server {
12.         listen 80;
13.         server_name localhost;
15.         client_max_body_size 100M;
17.         location / {
18.             proxy_pass http://api;
19.             proxy_set_header Host $host;
20.             proxy_set_header X-Real-IP $remote_addr;
21.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
22.             proxy_set_header X-Forwarded-Proto $scheme;
23.             proxy_read_timeout 300s;
24.             proxy_connect_timeout 75s;
25.         }
27.         location /docs {
28.             proxy_pass http://api;
29.             proxy_set_header Host $host;
30.             proxy_set_header X-Real-IP $remote_addr;
31.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
32.             proxy_set_header X-Forwarded-Proto $scheme;
33.         }
35.         location /redoc {
36.             proxy_pass http://api;
37.             proxy_set_header Host $host;
38.             proxy_set_header X-Real-IP $remote_addr;
39.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
40.             proxy_set_header X-Forwarded-Proto $scheme;
41.         }
43.         location /openapi.json {
44.             proxy_pass http://api;
45.             proxy_set_header Host $host;
46.             proxy_set_header X-Real-IP $remote_addr;
47.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
48.             proxy_set_header X-Forwarded-Proto $scheme;
49.         }
50.     }
51. }

复制代码

使用Docker Compose启动服务：

2. docker-compose up -d

复制代码

8.3 使用Kubernetes部署

创建一个deployment.yaml文件：

2. apiVersion: apps/v1
3. kind: Deployment
4. metadata:
5.   name: image-processing-api
6. spec:
7.   replicas: 3
8.   selector:
9.     matchLabels:
10.       app: image-processing-api
11.   template:
12.     metadata:
13.       labels:
14.         app: image-processing-api
15.     spec:
16.       containers:
17.       - name: api
18.         image: fastapi-image-processing:latest
19.         ports:
20.         - containerPort: 8000
21.         env:
22.         - name: ENVIRONMENT
23.           value: "production"
24.         volumeMounts:
25.         - name: temp-images
26.           mountPath: /app/temp_images
27.         - name: cache
28.           mountPath: /app/cache
29.         resources:
30.           requests:
31.             memory: "256Mi"
32.             cpu: "250m"
33.           limits:
34.             memory: "512Mi"
35.             cpu: "500m"
36.         livenessProbe:
37.           httpGet:
38.             path: /health
39.             port: 8000
40.           initialDelaySeconds: 30
41.           periodSeconds: 10
42.         readinessProbe:
43.           httpGet:
44.             path: /health
45.             port: 8000
46.           initialDelaySeconds: 5
47.           periodSeconds: 5
48.       volumes:
49.       - name: temp-images
50.         emptyDir: {}
51.       - name: cache
52.         emptyDir: {}
53. ---
54. apiVersion: v1
55. kind: Service
56. metadata:
57.   name: image-processing-api-service
58. spec:
59.   selector:
60.     app: image-processing-api
61.   ports:
62.     - protocol: TCP
63.       port: 80
64.       targetPort: 8000
65.   type: LoadBalancer
66. ---
67. apiVersion: networking.k8s.io/v1
68. kind: Ingress
69. metadata:
70.   name: image-processing-api-ingress
71.   annotations:
72.     nginx.ingress.kubernetes.io/rewrite-target: /
73.     nginx.ingress.kubernetes.io/proxy-body-size: "100m"
74. spec:
75.   rules:
76.   - http:
77.       paths:
78.       - path: /
79.         pathType: Prefix
80.         backend:
81.           service:
82.             name: image-processing-api-service
83.             port:
84.               number: 80

复制代码

部署到Kubernetes集群：

2. kubectl apply -f deployment.yaml

复制代码

9. 常见问题和解决方案

9.1 内存管理问题

问题：处理大图像时，可能会遇到内存不足的问题。

解决方案：

1. 限制上传文件大小：

2. from fastapi import Request, HTTPException
3. from fastapi.middleware import Middleware
4. from fastapi.middleware.base import BaseHTTPMiddleware
6. class MaxFileSizeMiddleware(BaseHTTPMiddleware):
7.     def __init__(self, app, max_file_size: int = 10 * 1024 * 1024):
8.         super().__init__(app)
9.         self.max_file_size = max_file_size
11.     async def dispatch(self, request: Request, call_next):
12.         content_type = request.headers.get("content-type", "")
13.         if "multipart/form-data" in content_type:
14.             content_length = request.headers.get("content-length")
15.             if content_length and int(content_length) > self.max_file_size:
16.                 raise HTTPException(status_code=413, detail="文件太大")
17.         response = await call_next(request)
18.         return response
20. # 添加中间件
21. app = FastAPI(
22.     title="图像处理API",
23.     description="使用FastAPI和Pillow构建的高性能图像处理API",
24.     middleware=[Middleware(MaxFileSizeMiddleware, max_file_size=10 * 1024 * 1024)]  # 10MB
25. )

复制代码

1. 使用流式处理：

2. @app.post("/process-stream/")
3. async def process_image_stream(
4.     file: UploadFile = File(...),
5.     operation: str = Form(...),
6.     # 其他参数...
7. ):
8.     """流式处理图像"""
9.     # 检查文件类型
10.     if not file.content_type.startswith("image/"):
11.         raise HTTPException(status_code=400, detail="文件必须是图像")
12.    
13.     try:
14.         # 使用流式读取
15.         contents = await file.read()
16.         
17.         # 处理图像...
18.         
19.         # 返回流式响应
20.         return StreamingResponse(
21.             io.BytesIO(processed_image_bytes),
22.             media_type=f"image/{output_format.lower()}",
23.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
24.         )
25.    
26.     except Exception as e:
27.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

1. 使用临时文件处理大图像：

2. import tempfile
3. import os
5. @app.post("/process-large/")
6. async def process_large_image(
7.     file: UploadFile = File(...),
8.     operation: str = Form(...),
9.     # 其他参数...
10. ):
11.     """处理大图像"""
12.     # 检查文件类型
13.     if not file.content_type.startswith("image/"):
14.         raise HTTPException(status_code=400, detail="文件必须是图像")
15.    
16.     try:
17.         # 创建临时文件
18.         with tempfile.NamedTemporaryFile(delete=False) as temp_file:
19.             # 流式写入临时文件
20.             for chunk in file.file:
21.                 temp_file.write(chunk)
22.             temp_file_path = temp_file.name
23.         
24.         try:
25.             # 使用临时文件处理图像
26.             image = Image.open(temp_file_path)
27.             
28.             # 处理图像...
29.             
30.             # 保存到另一个临时文件
31.             with tempfile.NamedTemporaryFile(suffix=f".{output_format.lower()}", delete=False) as output_temp_file:
32.                 image.save(output_temp_file.name, format=output_format, quality=quality)
33.                 output_temp_file_path = output_temp_file.name
34.             
35.             # 返回文件响应
36.             return FileResponse(
37.                 output_temp_file_path,
38.                 media_type=f"image/{output_format.lower()}",
39.                 filename=f"processed.{output_format.lower()}"
40.             )
41.         
42.         finally:
43.             # 清理临时文件
44.             if os.path.exists(temp_file_path):
45.                 os.unlink(temp_file_path)
46.    
47.     except Exception as e:
48.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

9.2 并发处理问题

问题：当多个用户同时上传大图像时，可能会导致服务器负载过高。

解决方案：

1. 使用队列系统处理任务：

2. from fastapi import BackgroundTasks
3. import redis
4. import json
5. import uuid
7. # Redis配置
8. REDIS_HOST = "localhost"
9. REDIS_PORT = 6379
10. REDIS_DB = 0
11. REDIS_QUEUE = "image_processing_tasks"
13. # 创建Redis连接
14. redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
16. @app.post("/process-queued/")
17. async def process_image_queued(
18.     file: UploadFile = File(...),
19.     operation: str = Form(...),
20.     # 其他参数...
21. ):
22.     """队列处理图像"""
23.     # 检查文件类型
24.     if not file.content_type.startswith("image/"):
25.         raise HTTPException(status_code=400, detail="文件必须是图像")
26.    
27.     # 创建任务ID
28.     task_id = str(uuid.uuid4())
29.    
30.     # 读取文件
31.     image_bytes = await file.read()
32.    
33.     # 准备任务数据
34.     task_data = {
35.         "task_id": task_id,
36.         "image_bytes": image_bytes.hex(),  # 转换为十六进制字符串
37.         "operation": operation,
38.         "width": width,
39.         "height": height,
40.         # 其他参数...
41.     }
42.    
43.     # 将任务添加到队列
44.     redis_client.rpush(REDIS_QUEUE, json.dumps(task_data))
45.    
46.     # 返回任务ID
47.     return {"task_id": task_id, "status": "queued", "message": "图像处理已加入队列"}
49. # 创建一个单独的工作进程来处理队列中的任务
50. def process_queue():
51.     """处理队列中的任务"""
52.     while True:
53.         # 从队列中获取任务
54.         task_json = redis_client.blpop(REDIS_QUEUE, timeout=1)
55.         if not task_json:
56.             continue
57.         
58.         try:
59.             # 解析任务数据
60.             task_data = json.loads(task_json[1])
61.             task_id = task_data["task_id"]
62.             
63.             # 更新任务状态
64.             redis_client.hset(f"task:{task_id}", "status", "processing")
65.             
66.             # 获取图像数据
67.             image_bytes = bytes.fromhex(task_data["image_bytes"])
68.             
69.             # 处理图像...
70.             
71.             # 更新任务状态
72.             redis_client.hset(f"task:{task_id}", "status", "completed")
73.             redis_client.hset(f"task:{task_id}", "result_url", f"/results/{task_id}.{output_format.lower()}")
74.             
75.         except Exception as e:
76.             # 更新任务状态为错误
77.             redis_client.hset(f"task:{task_id}", "status", "error")
78.             redis_client.hset(f"task:{task_id}", "message", str(e))
80. # 启动工作进程
81. import threading
82. worker_thread = threading.Thread(target=process_queue, daemon=True)
83. worker_thread.start()

复制代码

1. 使用Celery处理异步任务：

首先，安装Celery：

2. pip install celery[redis]

复制代码

创建一个celery_config.py文件：

2. from celery import Celery
3. import os
5. # Celery配置
6. CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0")
7. CELERY_RESULT_BACKEND = os.environ.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/0")
9. # 创建Celery应用
10. celery_app = Celery(
11.     "image_processing",
12.     broker=CELERY_BROKER_URL,
13.     backend=CELERY_RESULT_BACKEND,
14.     include=["tasks"]
15. )
17. # Celery配置
18. celery_app.conf.update(
19.     task_serializer="json",
20.     result_serializer="json",
21.     accept_content=["json"],
22.     timezone="UTC",
23.     enable_utc=True,
24.     task_track_started=True,
25.     task_time_limit=30 * 60,  # 30分钟超时
26.     worker_prefetch_multiplier=1,
27.     worker_max_tasks_per_child=1000,
28. )

复制代码

创建一个tasks.py文件：

2. from celery_config import celery_app
3. from PIL import Image, ImageFilter, ImageEnhance
4. import io
5. import os
6. from typing import Tuple, Optional
8. @celery_app.task(bind=True)
9. def process_image_task(self, image_bytes: bytes, operation: str, params: dict):
10.     """处理图像任务"""
11.     try:
12.         # 更新任务状态
13.         self.update_state(state="PROGRESS", meta={"step": "opening_image"})
14.         
15.         # 打开图像
16.         image = Image.open(io.BytesIO(image_bytes))
17.         
18.         # 执行操作
19.         if operation == "resize":
20.             self.update_state(state="PROGRESS", meta={"step": "resizing_image"})
21.             width = params.get("width")
22.             height = params.get("height")
23.             if width is None or height is None:
24.                 raise ValueError("调整大小需要宽度和高度参数")
25.             image = image.resize((width, height), Image.LANCZOS)
26.         elif operation == "crop":
27.             self.update_state(state="PROGRESS", meta={"step": "cropping_image"})
28.             left = params.get("left")
29.             top = params.get("top")
30.             right = params.get("right")
31.             bottom = params.get("bottom")
32.             if left is None or top is None or right is None or bottom is None:
33.                 raise ValueError("裁剪需要left、top、right和bottom参数")
34.             image = image.crop((left, top, right, bottom))
35.         # 其他操作...
36.         
37.         # 转换格式
38.         output_format = params.get("output_format", "JPEG")
39.         if output_format:
40.             self.update_state(state="PROGRESS", meta={"step": "converting_format"})
41.             if output_format.upper() == "JPEG" and image.mode in ("RGBA", "LA", "P"):
42.                 # JPEG不支持透明度，转换为RGB
43.                 background = Image.new("RGB", image.size, (255, 255, 255))
44.                 background.paste(image, mask=image.split()[-1] if image.mode == "RGBA" else None)
45.                 image = background
46.             elif output_format.upper() == "PNG" and image.mode not in ("RGBA", "LA", "P"):
47.                 # PNG支持透明度，转换为RGBA
48.                 image = image.convert("RGBA")
49.         
50.         # 保存图像
51.         self.update_state(state="PROGRESS", meta={"step": "saving_image"})
52.         quality = params.get("quality", 85)
53.         img_byte_arr = io.BytesIO()
54.         image.save(img_byte_arr, format=output_format, quality=quality)
55.         img_byte_arr.seek(0)
56.         processed_image_bytes = img_byte_arr.getvalue()
57.         
58.         # 返回结果
59.         return {
60.             "status": "completed",
61.             "result": processed_image_bytes.hex(),
62.             "format": output_format,
63.             "size": image.size
64.         }
65.    
66.     except Exception as e:
67.         # 更新任务状态为错误
68.         self.update_state(
69.             state="FAILURE",
70.             meta={
71.                 "error": str(e),
72.                 "status": "error"
73.             }
74.         )
75.         raise

复制代码

在main.py中添加使用Celery的端点：

2. from fastapi import FastAPI, File, UploadFile, HTTPException, Form
3. from fastapi.responses import StreamingResponse
4. from tasks import process_image_task
5. import io
7. @app.post("/process-celery/")
8. async def process_image_with_celery(
9.     file: UploadFile = File(...),
10.     operation: str = Form(...),
11.     # 其他参数...
12. ):
13.     """使用Celery处理图像"""
14.     # 检查文件类型
15.     if not file.content_type.startswith("image/"):
16.         raise HTTPException(status_code=400, detail="文件必须是图像")
17.    
18.     try:
19.         # 读取图像
20.         image_bytes = await file.read()
21.         
22.         # 准备参数
23.         params = {
24.             "width": width,
25.             "height": height,
26.             "angle": angle,
27.             "flip_mode": flip_mode,
28.             "filter_type": filter_type,
29.             "brightness": brightness,
30.             "contrast": contrast,
31.             "saturation": saturation,
32.             "output_format": output_format,
33.             "quality": quality,
34.             "left": left,
35.             "top": top,
36.             "right": right,
37.             "bottom": bottom
38.         }
39.         
40.         # 提交任务
41.         task = process_image_task.delay(image_bytes, operation, params)
42.         
43.         # 返回任务ID
44.         return {"task_id": task.id, "status": "queued", "message": "图像处理已开始"}
45.    
46.     except Exception as e:
47.         raise HTTPException(status_code=500, detail=f"提交任务时出错: {str(e)}")
49. @app.get("/celery-status/{task_id}")
50. async def get_celery_task_status(task_id: str):
51.     """获取Celery任务状态"""
52.     task = process_image_task.AsyncResult(task_id)
53.    
54.     if task.state == "PENDING":
55.         response = {
56.             "task_id": task_id,
57.             "status": "pending",
58.             "message": "任务等待中"
59.         }
60.     elif task.state == "PROGRESS":
61.         response = {
62.             "task_id": task_id,
63.             "status": "progress",
64.             "message": "任务处理中",
65.             "step": task.info.get("step", "")
66.         }
67.     elif task.state == "SUCCESS":
68.         response = {
69.             "task_id": task_id,
70.             "status": "completed",
71.             "message": "任务完成",
72.             "result_url": f"/celery-results/{task_id}"
73.         }
74.     elif task.state == "FAILURE":
75.         response = {
76.             "task_id": task_id,
77.             "status": "error",
78.             "message": "任务失败",
79.             "error": str(task.info)
80.         }
81.     else:
82.         response = {
83.             "task_id": task_id,
84.             "status": "unknown",
85.             "message": "未知状态"
86.         }
87.    
88.     return response
90. @app.get("/celery-results/{task_id}")
91. async def get_celery_task_result(task_id: str):
92.     """获取Celery任务结果"""
93.     task = process_image_task.AsyncResult(task_id)
94.    
95.     if task.state != "SUCCESS":
96.         raise HTTPException(status_code=404, detail="任务未完成或不存在")
97.    
98.     # 获取结果
99.     result = task.result
100.     processed_image_bytes = bytes.fromhex(result["result"])
101.     output_format = result["format"]
102.    
103.     # 返回处理后的图像
104.     return StreamingResponse(
105.         io.BytesIO(processed_image_bytes),
106.         media_type=f"image/{output_format.lower()}",
107.         headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
108.     )

复制代码

9.3 安全问题

问题：API可能会被恶意使用，如上传恶意文件、DDoS攻击等。

解决方案：

1. 文件类型验证：

2. from fastapi import UploadFile, HTTPException
3. import imghdr
5. def validate_image_file(file: UploadFile) -> bool:
6.     """验证图像文件"""
7.     # 检查MIME类型
8.     if not file.content_type.startswith("image/"):
9.         return False
10.    
11.     # 检查文件扩展名
12.     allowed_extensions = [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp"]
13.     file_extension = os.path.splitext(file.filename)[1].lower()
14.     if file_extension not in allowed_extensions:
15.         return False
16.    
17.     # 检查文件内容
18.     file_content = file.file.read()
19.     file.file.seek(0)  # 重置文件指针
20.    
21.     # 使用imghdr检查文件类型
22.     image_type = imghdr.what(None, h=file_content)
23.     if image_type not in ["jpeg", "jpg", "png", "gif", "bmp", "webp"]:
24.         return False
25.    
26.     return True
28. @app.post("/process-secure/")
29. async def process_image_secure(
30.     file: UploadFile = File(...),
31.     operation: str = Form(...),
32.     # 其他参数...
33. ):
34.     """安全处理图像"""
35.     # 验证文件
36.     if not validate_image_file(file):
37.         raise HTTPException(status_code=400, detail="无效的图像文件")
38.    
39.     try:
40.         # 处理图像...
41.         
42.         # 返回处理后的图像
43.         return StreamingResponse(
44.             io.BytesIO(processed_image_bytes),
45.             media_type=f"image/{output_format.lower()}",
46.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
47.         )
48.    
49.     except Exception as e:
50.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

1. 速率限制：

2. pip install fastapi-limiter

复制代码

2. from fastapi import FastAPI, Request
3. from fastapi_limiter import FastAPILimiter
4. from fastapi_limiter.depends import RateLimiter
5. import redis
7. app = FastAPI()
9. @app.on_event("startup")
10. async def startup():
11.     redis = await aioredis.create_redis_pool("redis://localhost")
12.     FastAPILimiter.init(redis)
14. @app.post("/process-rate-limited/", dependencies=[Depends(RateLimiter(times=10, seconds=60))])
15. async def process_image_rate_limited(
16.     file: UploadFile = File(...),
17.     operation: str = Form(...),
18.     # 其他参数...
19. ):
20.     """速率限制的图像处理"""
21.     # 验证文件
22.     if not validate_image_file(file):
23.         raise HTTPException(status_code=400, detail="无效的图像文件")
24.    
25.     try:
26.         # 处理图像...
27.         
28.         # 返回处理后的图像
29.         return StreamingResponse(
30.             io.BytesIO(processed_image_bytes),
31.             media_type=f"image/{output_format.lower()}",
32.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
33.         )
34.    
35.     except Exception as e:
36.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

1. 身份验证和授权：

2. from fastapi import Depends, HTTPException, status
3. from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
4. from jose import JWTError, jwt
5. from passlib.context import CryptContext
6. from datetime import datetime, timedelta
7. from pydantic import BaseModel
9. # 安全配置
10. SECRET_KEY = "your-secret-key-here"
11. ALGORITHM = "HS256"
12. ACCESS_TOKEN_EXPIRE_MINUTES = 30
14. # 密码上下文
15. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
17. # Bearer token
18. security = HTTPBearer()
20. # 用户模型
21. class User(BaseModel):
22.     username: str
23.     email: str = None
24.     full_name: str = None
25.     disabled: bool = None
27. class UserInDB(User):
28.     hashed_password: str
30. # 模拟数据库
31. fake_users_db = {
32.     "johndoe": {
33.         "username": "johndoe",
34.         "full_name": "John Doe",
35.         "email": "johndoe@example.com",
36.         "hashed_password": pwd_context.hash("secret"),
37.         "disabled": False,
38.     }
39. }
41. # Token模型
42. class Token(BaseModel):
43.     access_token: str
44.     token_type: str
46. class TokenData(BaseModel):
47.     username: str = None
49. def verify_password(plain_password, hashed_password):
50.     """验证密码"""
51.     return pwd_context.verify(plain_password, hashed_password)
53. def get_password_hash(password):
54.     """获取密码哈希"""
55.     return pwd_context.hash(password)
57. def get_user(db, username: str):
58.     """获取用户"""
59.     if username in db:
60.         user_dict = db[username]
61.         return UserInDB(**user_dict)
63. def authenticate_user(fake_db, username: str, password: str):
64.     """验证用户"""
65.     user = get_user(fake_db, username)
66.     if not user:
67.         return False
68.     if not verify_password(password, user.hashed_password):
69.         return False
70.     return user
72. def create_access_token(data: dict, expires_delta: timedelta = None):
73.     """创建访问令牌"""
74.     to_encode = data.copy()
75.     if expires_delta:
76.         expire = datetime.utcnow() + expires_delta
77.     else:
78.         expire = datetime.utcnow() + timedelta(minutes=15)
79.     to_encode.update({"exp": expire})
80.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
81.     return encoded_jwt
83. async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
84.     """获取当前用户"""
85.     credentials_exception = HTTPException(
86.         status_code=status.HTTP_401_UNAUTHORIZED,
87.         detail="无法验证凭据",
88.         headers={"WWW-Authenticate": "Bearer"},
89.     )
90.     try:
91.         payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
92.         username: str = payload.get("sub")
93.         if username is None:
94.             raise credentials_exception
95.         token_data = TokenData(username=username)
96.     except JWTError:
97.         raise credentials_exception
98.     user = get_user(fake_users_db, username=token_data.username)
99.     if user is None:
100.         raise credentials_exception
101.     return user
103. async def get_current_active_user(current_user: UserInDB = Depends(get_current_user)):
104.     """获取当前活跃用户"""
105.     if current_user.disabled:
106.         raise HTTPException(status_code=400, detail="禁用的用户")
107.     return current_user
109. @app.post("/token", response_model=Token)
110. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
111.     """登录获取访问令牌"""
112.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
113.     if not user:
114.         raise HTTPException(
115.             status_code=status.HTTP_401_UNAUTHORIZED,
116.             detail="用户名或密码不正确",
117.             headers={"WWW-Authenticate": "Bearer"},
118.         )
119.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
120.     access_token = create_access_token(
121.         data={"sub": user.username}, expires_delta=access_token_expires
122.     )
123.     return {"access_token": access_token, "token_type": "bearer"}
125. @app.post("/process-auth/")
126. async def process_image_auth(
127.     current_user: UserInDB = Depends(get_current_active_user),
128.     file: UploadFile = File(...),
129.     operation: str = Form(...),
130.     # 其他参数...
131. ):
132.     """需要身份验证的图像处理"""
133.     # 验证文件
134.     if not validate_image_file(file):
135.         raise HTTPException(status_code=400, detail="无效的图像文件")
136.    
137.     try:
138.         # 处理图像...
139.         
140.         # 返回处理后的图像
141.         return StreamingResponse(
142.             io.BytesIO(processed_image_bytes),
143.             media_type=f"image/{output_format.lower()}",
144.             headers={"Content-Disposition": f"attachment; filename=processed.{output_format.lower()}"}
145.         )
146.    
147.     except Exception as e:
148.         raise HTTPException(status_code=500, detail=f"处理图像时出错: {str(e)}")

复制代码

10. 结论

在本指南中，我们学习了如何使用FastAPI和Pillow构建高性能的图像处理API。我们从基础开始，逐步添加了更多功能，并解决了实际开发中可能遇到的各种挑战。

10.1 主要收获

1. FastAPI基础：我们学习了如何创建基本的FastAPI应用，添加路由，处理请求和响应。
2. Pillow图像处理：我们掌握了使用Pillow进行各种图像处理操作，如调整大小、裁剪、旋转、应用滤镜等。
3. API设计：我们设计了RESTful API，支持多种图像处理操作，并提供了批量处理功能。
4. 性能优化：我们实现了异步处理、缓存机制和队列系统，以提高API的性能和可扩展性。
5. 部署：我们学习了如何使用Docker、Docker Compose和Kubernetes部署API。
6. 问题解决：我们解决了实际开发中可能遇到的各种问题，如内存管理、并发处理和安全问题。

FastAPI基础：我们学习了如何创建基本的FastAPI应用，添加路由，处理请求和响应。

Pillow图像处理：我们掌握了使用Pillow进行各种图像处理操作，如调整大小、裁剪、旋转、应用滤镜等。

API设计：我们设计了RESTful API，支持多种图像处理操作，并提供了批量处理功能。

性能优化：我们实现了异步处理、缓存机制和队列系统，以提高API的性能和可扩展性。

部署：我们学习了如何使用Docker、Docker Compose和Kubernetes部署API。

问题解决：我们解决了实际开发中可能遇到的各种问题，如内存管理、并发处理和安全问题。

10.2 进一步学习

如果你想进一步学习，可以探索以下主题：

1. 更多图像处理功能：学习更高级的图像处理技术，如人脸识别、对象检测、图像分割等。
2. 机器学习集成：将机器学习模型集成到API中，以实现更智能的图像处理。
3. 微服务架构：将API拆分为多个微服务，以提高可维护性和可扩展性。
4. 监控和日志：添加监控和日志系统，以跟踪API的性能和使用情况。
5. CI/CD：实现持续集成和持续部署，以自动化测试和部署过程。

更多图像处理功能：学习更高级的图像处理技术，如人脸识别、对象检测、图像分割等。

机器学习集成：将机器学习模型集成到API中，以实现更智能的图像处理。

微服务架构：将API拆分为多个微服务，以提高可维护性和可扩展性。

监控和日志：添加监控和日志系统，以跟踪API的性能和使用情况。

CI/CD：实现持续集成和持续部署，以自动化测试和部署过程。

10.3 资源

以下是一些有用的资源，可以帮助你进一步学习：

1. FastAPI官方文档
2. Pillow官方文档
3. Docker官方文档
4. Kubernetes官方文档
5. Celery官方文档

希望本指南能帮助你构建高性能的图像处理API，并在实际开发中解决各种挑战。祝你编码愉快！