很多人写 Python 都有这个体验:本地脚本跑得好好的,自己用着很爽,结果一问同事要不要一起用,大家面面相觑——“你把脚本发我?”、“环境怎么装?”、“这个参数是啥?”…… 最简单的办法,其实就是:把它变成一个 HTTP API,谁想用就发个请求就行了。
下面我就用尽量“能落地”的方式,带着你从一个普通的 Python 函数,走到一个可以被别人调用的 API 服务,尽量不说废话,有代码、有命令,跟着敲就能跑起来。
先假装你现在有个老实巴交的函数,就放在 calc.py 里:
# calc.pydefadd(a: float, b: float) -> float:return a + bdefslow_square(x: float) -> float:""" 假装这是个很耗时的计算,比如调了模型、查了数据库等等 """import time time.sleep(1)return x * x现在问题是:
总不能每个人都拉一份代码+配置环境吧,这时候就该 API 上场了。
先来一个最朴素的方式,Flask,轻便好上手。
1)先装依赖:
pip install flask2)新建一个 app_flask.py:
# app_flask.pyfrom flask import Flask, request, jsonifyfrom calc import add, slow_squareapp = Flask(__name__)@app.route("/add", methods=["GET"])defadd_api():# 从查询参数里拿 a、b,比如 /add?a=1&b=2try: 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"])defsquare_api():# 从 JSON 里拿 x,比如 {"x": 3} data = request.get_json(silent=True) or {}if"x"notin data:return jsonify({"error": "缺少参数 x"}), 400try: 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)3)跑起来:
python app_flask.py4)怎么调用?
浏览器直接访问: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}'requests:import requestsresp = 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 请求。
Flask 虽然简单,但参数校验、自动文档就得你自己搞。 想要“写一点类型注解就自动生成接口文档”的,可以上 FastAPI。
1)安装依赖:
pip install fastapi uvicorn[standard]2)写个 app_fastapi.py:
# app_fastapi.pyfrom fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom calc import add, slow_squareapp = FastAPI(title="Demo Calculator API")classAddQuery(BaseModel): a: float b: floatclassSquareBody(BaseModel): x: float@app.get("/add")defadd_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")defsquare_endpoint(body: SquareBody):try: result = slow_square(body.x)except Exception as e:raise HTTPException(status_code=500, detail=str(e))return {"result": result}3)启动服务(注意这次用的是 uvicorn):
uvicorn app_fastapi:app --host 0.0.0.0 --port 8000 --reload4)有个很香的小功能:打开http://127.0.0.1:8000/docs你会看到自动生成的 Swagger UI,可以在页面里点点按钮、填个参数就调接口,非常适合甩给前端 or 测试同学。
同样地,前端或者其他服务,只要会发 HTTP 请求就能用你的逻辑了。
上面的例子是从“函数”到“API”。 真实项目常见的情况是:你已经有一大堆业务逻辑,可能还封在类里、模块里,改起来不想动太多代码。
一个建议是:
比如你原来有个 service.py:
# service.pyclassPriceService:def__init__(self, tax_rate: float = 0.1): self.tax_rate = tax_ratedefcalc_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.pyfrom fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModel, Fieldfrom service import PriceServiceapp = FastAPI(title="Price Service API")price_service = PriceService(tax_rate=0.13) # 比如税率写死或从配置加载classPriceRequest(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")defcalc_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}这样做的好处是:
service.py本地 python app.py 开发没问题,但要给别人用、特别是要上服务器,还得注意两点:
1)用专门的服务启动工具
示例(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 42)考虑 Docker 打包(这里只给个最简模板,有需要你再往下挖)
# DockerfileFROM 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 . /appEXPOSE8000CMD ["uvicorn", "app_fastapi:app", "--host", "0.0.0.0", "--port", "8000"]对应的 requirements.txt 可能就是:
fastapiuvicorn[standard]然后:
docker build -t my-python-api .docker run -p 8000:8000 my-python-api这样你的 Python 代码就完全变成一个“黑盒服务”,谁都可以通过 HTTP 调用。
不整一堆 checklist,就挑几个实战里经常踩的:
1)输入一定要校验
别相信客户端传的永远是好数据。
比如上面 Flask 的那段,其实已经是“最基础的校验”了。
2)错误别都返回 200
常见两个极端:
200 + "success": false比较自然的做法是:
FastAPI 里用 HTTPException 就挺顺手:
from fastapi import HTTPExceptionif something_wrong:raise HTTPException(status_code=400, detail="参数不合法")3)长耗时逻辑别直接让接口卡死
比如 slow_square 里我只是 sleep(1),可现实里可能是:
最简单的方式就是把“真正耗时的逻辑”丢到后台任务 / 消息队列里,这个就超出“快速变 API”的范围了,这里点一下就好。 先有最简单的同步版跑起来,把接口抽象清楚,再考虑异步、队列之类的优化。
最后拼一个非常完整又尽量短的小例子,你可以直接复制成一个目录跑起来:
目录结构:
my_api_demo/ calc.py service.py app.py requirements.txtcalc.py:
defadd(a: float, b: float) -> float:return a + bservice.py:
from calc import addclassCalcService:defadd_with_log(self, a: float, b: float) -> float:# 以后这里可以加日志、埋点、限流等等 result = add(a, b)return resultapp.py(用 FastAPI):
from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom service import CalcServiceapp = FastAPI(title="Mini Calc API")svc = CalcService()classAddReq(BaseModel): a: float b: float@app.post("/add")defadd_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:
fastapiuvicorn[standard]然后:
pip install -r requirements.txtuvicorn 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 函数 → 包一层 service → 暴露为 HTTP API → 本地/容器里都能跑”。
后面你想再加鉴权、限流、日志、监控、灰度发布之类,就在这个基础上慢慢往上盖楼就行了。
-END-
我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html
虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB