快速将Python业务函数封装为HTTP服务：Flask与FastAPI实战指南

你是否经常遇到这样的情况：辛苦写完的Python脚本，同事隔三差五就来问：“那个处理订单的脚本还能跑吗？帮忙跑一下，改两个参数就好。” 于是你不得不重复打开脚本、修改路径和参数、然后执行，效率低下且容易出错。

此时，一个高效的解决方案是将这段核心业务逻辑封装成一个HTTP API。这样一来，无论调用方是谁，只需要一个URL和简单的请求示例即可完成任务，无需关心背后的实现细节。

本文将以一种“最小化改造”的思路，展示如何将一段普通的Python脚本代码，逐步演变为一个可通过curl、前端或其他服务调用的健壮API。整个过程不涉及复杂的框架概念，先从最基础的脚本形态开始。

从脚本到API：核心思想

假设你原有的业务代码在一个utils.py文件中，其中包含一个用于统计订单金额的函数：

# utils.py
from typing import List, Dict

def summarize_orders(amounts: List[float]) -> Dict[str, float]:
    if not amounts:
        return {"count": 0, "sum": 0.0, "avg": 0.0, "max": 0.0, "min": 0.0}
    total = sum(amounts)
    return {
        "count": len(amounts),
        "sum": total,
        "avg": total / len(amounts),
        "max": max(amounts),
        "min": min(amounts),
    }

if __name__ == "__main__":
    # 以前的本地调用方式
    example = [10.5, 20, 30.2]
    print(summarize_orders(example))

这是我们常见的“可执行脚本”形态。改造的目标是：保持业务逻辑（summarize_orders函数）绝对不变，仅在其外部包裹一层HTTP API层。

方案一：使用Flask快速搭建API

对于追求快速上线的场景，Flask 是一个轻量且足够的选择。首先安装它：

pip install flask

然后，新建一个api_flask.py文件，作为API的入口：

# api_flask.py
from flask import Flask, request, jsonify
from utils import summarize_orders  # 直接导入原有的业务函数

app = Flask(__name__)

def error_response(code: str, message: str, http_status: int = 400):
    """统一错误响应格式"""
    return jsonify({
        "error_code": code,
        "message": message
    }), http_status

@app.route("/summarize", methods=["POST"])
def summarize_api():
    # 1. 解析HTTP请求体中的JSON数据
    data = request.get_json(silent=True) or {}

    # 2. 参数校验
    amounts = data.get("amounts")
    if not isinstance(amounts, list):
        return error_response("INVALID_PARAM", "amounts 必须是数组，例如 [10.5, 20, 30]")

    try:
        # 确保所有输入可转换为浮点数
        float_amounts = [float(x) for x in amounts]
    except (TypeError, ValueError):
        return error_response("INVALID_PARAM", "amounts 里的值必须能转换为数字")

    # 3. 调用核心业务逻辑（完全不变）
    summary = summarize_orders(float_amounts)

    # 4. 将Python字典封装为JSON HTTP响应
    return jsonify(summary)

if __name__ == "__main__":
    # 开发环境运行
    app.run(host="0.0.0.0", port=5000, debug=True)

启动服务：

python api_flask.py

使用curl测试API：

curl -X POST "http://127.0.0.1:5000/summarize" \
  -H "Content-Type: application/json" \
  -d '{"amounts": [12.5, 30, 7.8]}'

你将得到如下结构的JSON响应：

{
  "avg": 16.7666666667,
  "count": 3,
  "max": 30.0,
  "min": 7.8,
  "sum": 50.3
}

至此，你的Python函数已经成功转变为一个标准的HTTP API。

方案二：使用FastAPI获取更佳开发体验

如果你希望获得自动化的API文档、基于Pydantic的强类型校验以及更现代的异步支持，FastAPI 是更优的选择。

安装依赖：

pip install fastapi uvicorn

创建api_fastapi.py文件：

# api_fastapi.py
from typing import List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from utils import summarize_orders

app = FastAPI(title="订单统计API", description="将订单金额列表转换为统计摘要")

# 使用Pydantic模型定义请求和响应的数据结构
class SummarizeRequest(BaseModel):
    amounts: List[float]

class SummarizeResponse(BaseModel):
    count: int
    sum: float
    avg: float
    max: float
    min: float

@app.post("/summarize", response_model=SummarizeResponse)
def summarize(req: SummarizeRequest):
    # FastAPI会自动校验请求体是否符合SummarizeRequest模型
    if not req.amounts:
        raise HTTPException(status_code=400, detail="amounts 不能为空")

    # 调用不变的业务函数
    result = summarize_orders(req.amounts)
    return result

使用Uvicorn启动服务：

uvicorn api_fastapi:app --host 0.0.0.0 --port 8000 --reload

启动后，访问 http://127.0.0.1:8000/docs 即可看到自动生成的交互式Swagger文档。你可以直接在浏览器中尝试调用接口，极大地方便了前后端联调和API测试。

保持代码清晰：简单的分层设计

在将脚本转化为API的初期，应避免将所有逻辑堆砌在同一个文件。一个简单有效的分层原则是：

1. 业务层：utils.py中的函数，纯粹处理业务逻辑，对HTTP无感知。
2. API层：api_*.py文件，只负责三件事：
   • HTTP请求到Python参数的转换（如JSON解析）。
   • 调用业务层函数。
   • 将业务结果封装为HTTP响应。

这种分离使得业务逻辑可以在命令行工具、定时任务等其他场景中被复用，而无需从Web框架中剥离。

增强API的健壮性与安全性

脚本私下运行与作为API服务运行的关键区别在于“用户”。为了提供更可靠的服务，建议至少增加以下三项基础配置：

1. 统一错误处理：如上文代码所示，定义一致的错误响应格式（如包含error_code和message），方便调用方排查问题。

2. 基础日志记录：记录请求的路径、来源IP和关键参数，便于追踪和审计。

   # Flask日志示例
   import logging
   from flask import request
   logging.basicConfig(level=logging.INFO)
   logger = logging.getLogger(__name__)
   
   @app.before_request
   def log_request_info():
       logger.info(f"Path: {request.path}, IP: {request.remote_addr}, Args: {request.get_json(silent=True)}")

3. 简单的API密钥鉴权：对于内部接口，可通过校验请求头中的特定字段实现基础防护。

   VALID_API_KEYS = {"your-secure-key-here"}
   
   @app.before_request
   def check_auth():
       if request.endpoint != 'summarize_api':  # 可排除某些公开接口
           api_key = request.headers.get('X-API-Key')
           if api_key not in VALID_API_KEYS:
               return error_response("UNAUTHORIZED", "无效的API密钥", 401)

从开发到部署：几种常见路径

本地API调试完成后，需要考虑如何部署以供他人使用。

1. 传统服务器部署：使用Gunicorn等WSGI服务器替代Flask内置服务器，并通过Systemd等工具托管进程，保障服务稳定性。

   pip install gunicorn
   gunicorn -w 4 -b 0.0.0.0:5000 api_flask:app

2. 容器化部署：使用Docker将应用及其依赖打包成镜像，这是当前更主流的、环境一致的部署方式。

   # Dockerfile
   FROM python:3.11-slim
   WORKDIR /app
   COPY requirements.txt .
   RUN pip install -r requirements.txt
   COPY . .
   CMD ["uvicorn", "api_fastapi:app", "--host", "0.0.0.0", "--port", "8000"]

   构建并运行容器后，你的服务就具备了高度可移植性，可以轻松在云原生或Kubernetes环境中运行。

3. Serverless/云函数：对于调用频率不高的小工具型API，可以直接部署到各大云平台的函数计算服务（如AWS Lambda、阿里云函数计算），实现按需运行和免运维管理。

何时应该进行重构？

“快速实现”是项目启动期的优势，但当出现以下信号时，应考虑进行适度的代码重构，以避免长期维护成本上升：

• 接口数量增多，单个文件变得臃肿。
• 多个接口间出现大量重复的校验或格式化代码。
• 调用方增加，问题排查需要更清晰的日志和错误追踪。
  此时，即使只是简单地将通用工具函数（如错误处理、日志配置、数据库连接池）抽离到独立的common模块中，也能显著提升代码的可维护性。