# fastapi-wrapper > 将 Python 函数或模块封装为标准化的 FastAPI RESTful API 服务，包含统一响应、日志追踪、文档生成等功能 - Author: Duangang - Repository: CoderXiaopang/oh-my-skills - Version: 20260123093251 - Stars: 0 - Forks: 0 - Last Updated: 2026-02-06 - Source: https://github.com/CoderXiaopang/oh-my-skills - Web: https://mule.run/skillshub/@@CoderXiaopang/oh-my-skills~fastapi-wrapper:20260123093251 --- --- name: fastapi-wrapper description: 将 Python 函数或模块封装为标准化的 FastAPI RESTful API 服务，包含统一响应、日志追踪、文档生成等功能 --- # FastAPI 封装技能 ## 使用场景 当用户需要以下任一功能时使用此 skill： - 将现有 Python 函数/类封装为 HTTP API 接口 - 快速搭建带有日志追踪的 API 服务 - 生成标准化的 API 文档和调用示例 - 统一 API 响应格式和错误处理 ## 核心目标 将用户代码转换为包含以下特性的完整 FastAPI 项目： 1. **标准项目结构**：清晰的模块划分 2. **统一响应格式**：`{success, message, data, request_id}` 3. **请求追踪**：基于 `request_id` 的完整日志链路 4. **自动文档**：Swagger UI + API.md + 调用示例 5. **健壮性**：参数验证、异常处理、类型注解 ## 项目结构 ``` project/ ├── main.py # FastAPI 应用入口 ├── api/ │ ├── routes.py # API 路由 │ └── models.py # Pydantic 请求/响应模型 ├── core/ │ ├── config.py # 配置管理 │ └── logger.py # 日志系统 ├── services/ │ └── business.py # 业务逻辑封装 ├── utils/ │ ├── response.py # 统一响应工具 │ └── image_handler.py # 图片格式处理工具（支持 base64/URL/文件流） ├── docs/ │ ├── API.md # API 文档 │ └── example_call.py # Python 调用示例 ├── tests/ │ └── test_api.py # API 测试 ├── requirements.txt └── README.md ``` ## 封装流程 ### 1. 分析源代码 - 识别函数签名（参数类型、返回值、默认值） - 确定异常类型和处理策略 - 检查外部依赖 ### 2. 创建数据模型 (`api/models.py`) ```python from pydantic import BaseModel, Field class YourRequest(BaseModel): param1: str = Field(..., description="参数说明") param2: int = Field(default=0, ge=0) class Config: json_schema_extra = { "example": {"param1": "value", "param2": 10} } ``` ### 3. 统一响应格式 (`utils/response.py`) ```python from typing import Any, Optional from pydantic import BaseModel class ResponseModel(BaseModel): success: bool message: str data: Optional[Any] = None request_id: str def success_response(data, message="Success", request_id=""): return ResponseModel(success=True, message=message, data=data, request_id=request_id) def error_response(message, request_id=""): return ResponseModel(success=False, message=message, data=None, request_id=request_id) ``` ### 4. 日志系统集成 (`core/logger.py`) **关键要点**: - 使用 `ContextVar` 存储 `request_id` - 中间件自动生成/读取 `X-Request-ID` 请求头 - 所有日志包含 `request_id` 参考 `references/logging.md` 查看完整实现。 ### 5. 业务逻辑迁移 (`services/business.py`) 将用户原始代码迁移到此处，保持业务逻辑独立。 ### 6. 图片输入处理 (`utils/image_handler.py`) **适用场景**：当 API 需要接收图片作为输入时。 **核心目标**： - 支持多种输入格式：base64 编码、URL、文件上传（文件流）、本地路径 - 统一转换为内部使用的格式（如 PIL.Image、numpy.ndarray、文件路径等） - 提供清晰的错误提示 ```python import base64 import io import os import tempfile from typing import Union, Literal from pathlib import Path from PIL import Image import requests from fastapi import UploadFile from core.logger import get_logger logger = get_logger(__name__) class ImageInputHandler: """ 统一处理多种图片输入格式的工具类 支持：base64、URL、文件上传、本地路径 """ @staticmethod def to_pil_image(image_input: Union[str, UploadFile, bytes]) -> Image.Image: """ 将任意格式的图片输入转换为 PIL Image 对象 Args: image_input: 可以是 base64 字符串、URL、UploadFile 对象或字节数据 Returns: PIL.Image.Image 对象 Raises: ValueError: 无法识别或处理的输入格式 """ try: # 1. 处理 UploadFile (FastAPI 文件上传) if isinstance(image_input, UploadFile): logger.info(f"Processing uploaded file: {image_input.filename}") contents = image_input.file.read() return Image.open(io.BytesIO(contents)) # 2. 处理字节数据 elif isinstance(image_input, bytes): logger.info("Processing bytes data") return Image.open(io.BytesIO(image_input)) # 3. 处理字符串（base64 或 URL 或本地路径） elif isinstance(image_input, str): # 3.1 检查是否为 base64 编码 if image_input.startswith('data:image'): logger.info("Processing base64 data URL") # 格式: data:image/png;base64,iVBORw0KG... base64_data = image_input.split(',', 1)[1] image_bytes = base64.b64decode(base64_data) return Image.open(io.BytesIO(image_bytes)) elif len(image_input) > 100 and not any(c in image_input for c in ['/', ':', '.']): # 纯 base64 字符串（无前缀） logger.info("Processing pure base64 string") image_bytes = base64.b64decode(image_input) return Image.open(io.BytesIO(image_bytes)) # 3.2 检查是否为 URL elif image_input.startswith(('http://', 'https://')): logger.info(f"Downloading image from URL: {image_input}") response = requests.get(image_input, timeout=30) response.raise_for_status() return Image.open(io.BytesIO(response.content)) # 3.3 检查是否为本地文件路径 elif Path(image_input).exists(): logger.info(f"Loading image from local path: {image_input}") return Image.open(image_input) else: raise ValueError(f"无法识别的字符串输入格式: {image_input[:50]}...") else: raise ValueError(f"不支持的输入类型: {type(image_input)}") except Exception as e: logger.error(f"图片处理失败: {e}", exc_info=True) raise ValueError(f"图片处理失败: {str(e)}") @staticmethod def to_bytes(image_input: Union[str, UploadFile, bytes], format: str = "PNG") -> bytes: """ 将任意格式的图片输入转换为字节数据 Args: image_input: 任意图片输入格式 format: 输出图片格式 (PNG, JPEG 等) Returns: 图片的字节数据 """ pil_image = ImageInputHandler.to_pil_image(image_input) buffer = io.BytesIO() pil_image.save(buffer, format=format) return buffer.getvalue() @staticmethod def to_base64(image_input: Union[str, UploadFile, bytes], format: str = "PNG", include_prefix: bool = True) -> str: """ 将任意格式的图片输入转换为 base64 字符串 Args: image_input: 任意图片输入格式 format: 输出图片格式 include_prefix: 是否包含 data:image/xxx;base64, 前缀 Returns: base64 编码的字符串 """ image_bytes = ImageInputHandler.to_bytes(image_input, format) base64_str = base64.b64encode(image_bytes).decode('utf-8') if include_prefix: mime_type = f"image/{format.lower()}" return f"data:{mime_type};base64,{base64_str}" return base64_str @staticmethod def to_temp_file(image_input: Union[str, UploadFile, bytes], suffix: str = ".png") -> str: """ 将任意格式的图片输入保存为临时文件 Args: image_input: 任意图片输入格式 suffix: 临时文件后缀 Returns: 临时文件的绝对路径 """ pil_image = ImageInputHandler.to_pil_image(image_input) temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) pil_image.save(temp_file.name) logger.info(f"Saved to temp file: {temp_file.name}") return temp_file.name @staticmethod def to_numpy(image_input: Union[str, UploadFile, bytes]): """ 将任意格式的图片输入转换为 numpy 数组 (用于 CV 模型) Returns: numpy.ndarray """ import numpy as np pil_image = ImageInputHandler.to_pil_image(image_input) return np.array(pil_image) ``` **集成到 API 模型**： ```python # api/models.py from pydantic import BaseModel, Field from typing import Union, Optional from fastapi import UploadFile, File, Form class ImageRequest(BaseModel): """ 支持多种图片输入格式的请求模型 使用时根据实际需求选择一种方式 """ image_base64: Optional[str] = Field( None, description="Base64 编码的图片（可带 data:image/xxx;base64, 前缀）" ) image_url: Optional[str] = Field( None, description="图片的 HTTP/HTTPS URL" ) image_path: Optional[str] = Field( None, description="服务器本地图片路径（谨慎使用）" ) class Config: json_schema_extra = { "example": { "image_base64": "data:image/png;base64,iVBORw0KG...", "image_url": "https://example.com/image.jpg", "image_path": "/path/to/local/image.png" } } # 如果需要支持文件上传，使用 Form + File # 注意：文件上传不能与 JSON body 同时使用 ``` **在路由中使用**： ```python # api/routes.py - 方式1：JSON 传参（base64/URL/path） @router.post("/process-image") async def process_image(request: ImageRequest): try: # 统一获取图片输入 image_input = request.image_base64 or request.image_url or request.image_path if not image_input: return error_response("必须提供一种图片输入", get_request_id()) # 根据业务需求转换为对应格式 pil_image = ImageInputHandler.to_pil_image(image_input) # 调用业务逻辑 result = your_image_processing_function(pil_image) return success_response(data=result, request_id=get_request_id()) except Exception as e: logger.error(f"处理失败: {e}", exc_info=True) return error_response(str(e), get_request_id()) # api/routes.py - 方式2：文件上传 @router.post("/upload-image") async def upload_image(file: UploadFile = File(...)): try: # 直接传入 UploadFile 对象 pil_image = ImageInputHandler.to_pil_image(file) result = your_image_processing_function(pil_image) return success_response(data=result, request_id=get_request_id()) except Exception as e: logger.error(f"处理失败: {e}", exc_info=True) return error_response(str(e), get_request_id()) # api/routes.py - 方式3：混合模式（推荐） @router.post("/flexible-image") async def flexible_image( image_base64: Optional[str] = Form(None), image_url: Optional[str] = Form(None), file: Optional[UploadFile] = File(None) ): """ 最灵活的方式：同时支持表单中的 base64/URL 和文件上传 """ try: # 按优先级选择输入 if file: image_input = file elif image_base64: image_input = image_base64 elif image_url: image_input = image_url else: return error_response("必须提供图片输入", get_request_id()) pil_image = ImageInputHandler.to_pil_image(image_input) result = your_image_processing_function(pil_image) return success_response(data=result, request_id=get_request_id()) except Exception as e: return error_response(str(e), get_request_id()) ``` ### 7. API 路由定义 (`api/routes.py`) ```python from fastapi import APIRouter, Header from api.models import YourRequest from services.business import your_function from utils.response import success_response, error_response from core.logger import get_logger, get_request_id router = APIRouter(prefix="/api/v1") logger = get_logger(__name__) @router.post("/endpoint") async def endpoint(request: YourRequest): try: logger.info(f"Processing request: {request.dict()}") result = your_function(request.param1, request.param2) return success_response(data=result, request_id=get_request_id()) except ValueError as e: return error_response(f"Invalid input: {e}", get_request_id()) except Exception as e: logger.error(f"Error: {e}", exc_info=True) return error_response("Internal error", get_request_id()) ``` ### 8. 主应用配置 (`main.py`) ```python from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from api.routes import router from core.logger import setup_logging, set_request_id, get_logger setup_logging() logger = get_logger(__name__) app = FastAPI(title="Your API", version="1.0.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"]) @app.middleware("http") async def log_middleware(request: Request, call_next): request_id = request.headers.get("X-Request-ID") set_request_id(request_id) logger.info(f"{request.method} {request.url.path}") response = await call_next(request) response.headers["X-Request-ID"] = get_request_id() return response app.include_router(router) @app.get("/health") def health(): return {"status": "ok"} ``` ### 9. 生成文档 #### `docs/API.md` 包含以下部分： - **基本信息**：URL、版本、认证方式 - **通用说明**：请求头、响应格式 - **接口详情**：每个端点的参数表格、请求/响应示例、错误码 参考 `references/api-doc-template.md` 模板。 #### `docs/example_call.py` 提供可执行的 Python 调用示例： ```python import requests def call_api(param1, param2=0): response = requests.post( "http://localhost:8000/api/v1/endpoint", json={"param1": param1, "param2": param2}, headers={"X-Request-ID": "custom-id"} ) result = response.json() if result["success"]: print(f"✓ Success: {result['data']}") else: print(f"✗ Error: {result['message']}") return result ``` ### 10. 测试 (`tests/test_api.py`) ```python from fastapi.testclient import TestClient from main import app client = TestClient(app) def test_endpoint(): response = client.post("/api/v1/endpoint", json={"param1": "test"}) assert response.status_code == 200 assert response.json()["success"] is True ``` ### 11. 依赖管理 (`requirements.txt`) ```txt fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 requests==2.31.0 pytest==7.4.3 # 图片处理相关依赖（如果需要图片输入支持） Pillow==10.1.0 numpy==1.24.3 python-multipart==0.0.6 # 支持文件上传 ``` ## 输出检查清单 提交前确保： - [ ] 所有函数有类型注解和文档字符串 - [ ] 每个接口都有对应的 Pydantic 模型 - [ ] 日志系统集成完毕，request_id 正确传递 - [ ] 统一响应格式应用于所有端点 - [ ] `docs/API.md` 包含完整的接口说明和示例 - [ ] `docs/example_call.py` 可直接运行 - [ ] `README.md` 包含快速启动指南 - [ ] 代码通过基本测试 - [ ] 如涉及图片输入，已实现 `ImageInputHandler` 并支持 base64/URL/文件流 - [ ] 图片输入接口正确处理了所有输入格式并返回清晰错误提示 ## 最佳实践 1. **请求验证**：使用 Pydantic 的 `Field`、`validator` 实现完整验证 2. **异步处理**：I/O 密集型操作使用 `async def` 3. **依赖注入**：使用 `Depends` 管理认证、数据库连接等 4. **配置管理**：使用 `pydantic_settings.BaseSettings` + `.env` 5. **日志分级**：DEBUG（调试）、INFO（关键操作）、ERROR（异常） 6. **错误安全**：生产环境不要暴露详细错误堆栈 7. **图片输入处理**： - 始终使用 `ImageInputHandler` 统一处理，不要在业务代码中重复实现格式转换 - 接口层接收多格式输入，内部统一转换为业务逻辑需要的格式 - 对大文件设置合理的上传限制（在 `main.py` 中配置 `max_request_size`） - 下载外部 URL 图片时设置超时和大小限制，防止恶意攻击 - 及时清理临时文件，避免磁盘空间泄漏 ## 示例参考 查看 `assets/` 目录获取完整示例和自动化脚本： - `assets/example_call.py` - API 调用示例模板 - `assets/init_project.py` - 自动初始化项目结构的脚本 - `assets/image_api_example.py` - 图片处理 API 调用示例（支持 base64/URL/文件上传） --- **总结**：遵循此流程可快速将任意 Python 代码封装为生产级别的 FastAPI 服务。