云栈社区»论坛 › 技术文档「 Note & Doc 」 › 使用FastAPI与Flask将Python代码封装为API接口的实战指南 ...

发回帖发新帖

1835 积分	0 好友	226 主题

发消息

使用FastAPI与Flask将Python代码封装为API接口的实战指南

发表于 4 天前 | 查看: 13| 回复: 0

很多开发者都遇到过这样的场景：自己写了一个功能强大的 Python 脚本，在本地跑得顺风顺水。但当你想分享给同事，或者需要让前端、其他服务调用时，问题就来了——难道要让每个人都去配环境、看源码、传命令行参数吗？

其实，一个高效且通用的解决方案是：将你的脚本逻辑封装成一个 HTTP API 服务。这样，任何能发起网络请求的客户端，都可以方便地调用你的功能，实现逻辑的共享与复用。

下面，我将以“能落地”为原则，手把手带你从最普通的 Python 函数开始，一步步构建出可被外部调用的 API 服务。全程附代码和命令，跟着操作就能跑通。

从一个普通的 Python 函数开始

假设你有一个简单的计算模块，保存在 calc.py 文件中：

# calc.py

def add(a: float, b: float) -> float:
    return a + b

def slow_square(x: float) -> float:
    """
    模拟一个耗时操作，例如调用模型、查询复杂数据库等
    """
    import time
    time.sleep(1)
    return x * x

现在面临的问题是：

同事想要使用这个计算功能。
前端项目需要集成这个能力。
其他语言（如 Java、Go）编写的服务也需要调用。

显然，让每个人复制代码并配置环境是不现实的。此时，将其包装成 HTTP API 就成了最佳选择。

方案一：使用 Flask 快速搭建（五分钟极简版）

首先介绍最轻量、易上手的方式——使用 Flask 框架。

安装依赖：
```
pip install flask
```

创建 API 入口文件 app_flask.py：

# app_flask.py
from flask import Flask, request, jsonify
from calc import add, slow_square

app = Flask(__name__)

@app.route("/add", methods=["GET"])
def add_api():
    # 从URL查询参数获取 a 和 b，例如 /add?a=1&b=2
    try:
        a = float(request.args.get("a", "0"))
        b = float(request.args.get("b", "0"))
    except ValueError:
        return jsonify({"error": "a 或 b 不是数字"}), 400

    result = add(a, b)
    return jsonify({"result": result})

@app.route("/square", methods=["POST"])
def square_api():
    # 从JSON请求体中获取 x，例如 {"x": 3}
    data = request.get_json(silent=True) or {}
    if "x" not in data:
        return jsonify({"error": "缺少参数 x"}), 400

    try:
        x = float(data["x"])
    except (TypeError, ValueError):
        return jsonify({"error": "x 必须是数字"}), 400

    result = slow_square(x)
    return jsonify({"result": result})

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

启动服务：
```
python app_flask.py
```

如何调用？

浏览器直接访问：http://127.0.0.1:5000/add?a=1.5&b=2.5

使用 curl 调用 POST 接口：

curl -X POST "http://127.0.0.1:5000/square" \
     -H "Content-Type: application/json" \
     -d '{"x": 3}'

或在 Python 中使用 requests 库：

import requests

resp = requests.get(
    "http://127.0.0.1:5000/add",
    params={"a": 1.5, "b": 2.5}
)
print(resp.json())

resp2 = requests.post(
    "http://127.0.0.1:5000/square",
    json={"x": 3}
)
print(resp2.json())

至此，你已经成功将一个本地函数转换成了可通过网络调用的 API。调用方无需了解内部实现，只需遵循 HTTP 协议即可。

方案二：使用 FastAPI 获得更佳开发体验（自动文档与强类型）

Flask 足够简单，但参数校验、API 文档等需要额外工作。如果你追求“通过类型注解自动生成接口文档”的现代开发体验，FastAPI 是绝佳选择。

安装依赖：
```
pip install fastapi uvicorn[standard]
```

创建 app_fastapi.py：

# app_fastapi.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from calc import add, slow_square

app = FastAPI(title="Demo Calculator API")

class AddQuery(BaseModel):
    a: float
    b: float

class SquareBody(BaseModel):
    x: float

@app.get("/add")
def add_endpoint(a: float, b: float):
    # FastAPI 会自动进行类型转换和校验
    try:
        result = add(a, b)
    except Exception as e:
        # 实际项目中应记录日志
        raise HTTPException(status_code=500, detail=str(e))
    return {"result": result}

@app.post("/square")
def square_endpoint(body: SquareBody):
    try:
        result = slow_square(body.x)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {"result": result}

启动服务（使用 ASGI 服务器 uvicorn）：

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

FastAPI 的亮点功能：访问 http://127.0.0.1:8000/docs，你会看到自动生成的、交互式的 Swagger UI 文档。你可以直接在页面上尝试调用接口，这对于前后端联调和测试非常方便。

如何适配已有的复杂项目代码？

上面的例子演示了如何包装独立的函数。但在实际项目中，你的业务逻辑可能已经封装在类或复杂的模块中。这时，一个重要的原则是：保持 API 层的“薄”。

API 层应只负责：接收参数 → 调用核心业务逻辑 → 返回结果与处理异常。

例如，假设你已有一个 service.py，其中包含了复杂的业务类：

# service.py

class PriceService:

    def __init__(self, tax_rate: float = 0.1):
        self.tax_rate = tax_rate

    def calc_price(self, base_price: float, discount: float = 0.0) -> float:
        """
        返回含税后的价格
        """
        price = base_price * (1 - discount)
        price = price * (1 + self.tax_rate)
        return round(price, 2)

那么，你的 API 层（以 FastAPI 为例）可以这样写，专注于参数校验和请求响应：

# app_price.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from service import PriceService

app = FastAPI(title="Price Service API")
price_service = PriceService(tax_rate=0.13)  # 税率可写死或从配置加载

class PriceRequest(BaseModel):
    base_price: float = Field(..., gt=0, description="原价，必须大于 0")
    discount: float = Field(0.0, ge=0, le=0.9, description="折扣，0~0.9 之间")

@app.post("/price")
def calc_price_api(body: PriceRequest):
    try:
        result = price_service.calc_price(
            base_price=body.base_price,
            discount=body.discount,
        )
    except Exception as e:
        raise HTTPException(500, detail=f"内部错误：{e}")

    return {"price": result}

这种分层架构的好处显而易见：

业务逻辑与接口解耦，便于单独进行单元测试。
核心逻辑可复用，未来无论是通过 gRPC、命令行还是定时任务调用，都无需重写。
API 层职责清晰，只关注输入验证、协议适配和错误处理。

让 API 服务真正可用：简单部署指南

在本地用 python app.py 开发调试没问题，但要提供给他人使用或部署到服务器，还需要注意以下几点：

使用生产级服务器启动

Flask：建议使用 Gunicorn 配合 Gevent 等 Worker。
FastAPI：推荐使用 Uvicorn 或 Gunicorn + Uvicorn Worker。

以 FastAPI 为例，生产环境启动命令：

# 基础版本
uvicorn app_fastapi:app --host 0.0.0.0 --port 8000

# 启用多Worker，提升并发能力（适用于多核服务器）
uvicorn app_fastapi:app --host 0.0.0.0 --port 8000 --workers 4

使用 Docker 容器化部署（提供最简模板）
将应用打包成 Docker 镜像，可以极大简化环境依赖和部署流程。

Dockerfile:

# Dockerfile
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 复制依赖文件并安装
COPY requirements.txt /app/
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制应用代码
COPY . /app

# 声明容器暴露的端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "app_fastapi:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt:

fastapi
uvicorn[standard]

构建与运行:

docker build -t my-python-api .
docker run -p 8000:8000 my-python-api

通过Docker化，你的 Python 代码就彻底变成了一个独立的、开箱即用的“黑盒”服务。

开发 API 时容易忽略的几个要点

必须进行输入校验
永远不要信任客户端传来的数据。使用 FastAPI 时，其集成的 Pydantic 库能提供强大的类型校验。如果自己手写校验，至少要用 try/except 并判断必填字段。
合理使用 HTTP 状态码
常见的误区是所有响应都返回 200，然后在 body 里用 "success": false 表示错误；或者所有错误都直接抛 500。
更合理的做法是：
- 参数错误 → 400 Bad Request
- 未授权/无权限 → 401 Unauthorized / 403 Forbidden
- 资源不存在 → 404 Not Found
- 服务器内部错误 → 500 Internal Server Error
  FastAPI 中可使用 HTTPException 方便地抛出指定状态码的错误。
注意长耗时操作的阻塞问题
示例中的 slow_square 函数通过 sleep(1) 模拟了耗时操作。现实中，这可能是模型推理、调用第三方慢接口或处理大文件。
对于同步框架，长耗时请求会阻塞整个 Worker，影响并发能力。最简单的优化思路是将耗时任务异步化，例如放入后台任务队列（如 Celery）处理，并通过轮询或 WebSocket 通知客户端结果。在初期，可以先用同步版本跑通流程，后续再根据压力情况引入异步机制。

完整示例：从脚本到 API 的迷你项目

最后，我们整合一个完整的、可运行的迷你示例项目。

目录结构:

my_api_demo/
  calc.py
  service.py
  app.py
  requirements.txt

calc.py (核心计算函数):

def add(a: float, b: float) -> float:
    return a + b

service.py (业务逻辑层):

from calc import add

class CalcService:

    def add_with_log(self, a: float, b: float) -> float:
        # 这里未来可以方便地添加日志、埋点、限流等逻辑
        result = add(a, b)
        return result

app.py (API 层，使用 FastAPI):

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from service import CalcService

app = FastAPI(title="Mini Calc API")
svc = CalcService()

class AddReq(BaseModel):
    a: float
    b: float

@app.post("/add")
def add_api(body: AddReq):
    try:
        res = svc.add_with_log(body.a, body.b)
    except Exception as e:
        raise HTTPException(500, detail=f"内部错误: {e}")
    return {"result": res}

requirements.txt:

fastapi
uvicorn[standard]

运行与测试:

安装依赖：pip install -r requirements.txt
启动服务：uvicorn app:app --host 0.0.0.0 --port 8000 --reload

发起请求测试：

curl -X POST "http://127.0.0.1:8000/add" \
     -H "Content-Type: application/json" \
     -d '{"a": 10, "b": 20.5}'

预期返回：{"result": 30.5}

至此，你已经完整走通了 “编写 Python 函数 → 封装业务逻辑层 → 暴露为 HTTP API → 本地/容器化运行” 的全流程。在此基础上，后续可以根据需求逐步添加身份认证、请求限流、结构化日志、性能监控等功能。如果你在实践过程中有任何心得或疑问，也欢迎到云栈社区与更多开发者交流探讨。

上一篇：华山A2000芯片通过美国审查获准全球销售，加速高阶智能驾驶商业化
下一篇：MySQL查询优化器深度解析：为何索引可用却执行全表扫描的成本计算逻辑

Python, FastAPI, Flask, Docker, API服务

收藏0 回复显示全部楼层举报

返回列表