如何快速把你的 Python 代码变为 API
很多人写 Python 都有这个体验:本地脚本跑得好好的,自己用着很爽,结果一问同事要不要一起用,大家面面相觑——“你把脚本发我?”、“环境怎么装?”、“这个参数是啥?”…… 最简单的办法,其实就是:把它变成一个 HTTP API,谁想用就发个请求就行了。
下面我就用尽量“能落地”的方式,带着你从一个普通的 Python 函数,走到一个可以被别人调用的 API 服务,尽量不说废话,有代码、有命令,跟着敲就能跑起来。
先有一段“普通”的 Python 代码
先假装你现在有个老实巴交的函数,就放在 calc.py 里:
# calc.py
defadd(a: float, b: float) -> float:
return a + b
defslow_square(x: float) -> float:
"""
假装这是个很耗时的计算,比如调了模型、查了数据库等等
"""
import time
time.sleep(1)
return x * x
现在问题是:
你同事要用这个功能 前端要用这个功能 其他语言(比如 Java、Go)写的服务也要用
总不能每个人都拉一份代码+配置环境吧,这时候就该 API 上场了。
用 Flask 五分钟糙快版:能跑就行
先来一个最朴素的方式,Flask,轻便好上手。
1)先装依赖:
pip install flask
2)新建一个 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"])
defadd_api():
# 从查询参数里拿 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"])
defsquare_api():
# 从 JSON 里拿 x,比如 {"x": 3}
data = request.get_json(silent=True) or {}
if"x"notin 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)
3)跑起来:
python app_flask.py
4)怎么调用?
浏览器直接访问:
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 虽然简单,但参数校验、自动文档就得你自己搞。 想要“写一点类型注解就自动生成接口文档”的,可以上 FastAPI。
1)安装依赖:
pip install fastapi uvicorn[standard]
2)写个 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")
classAddQuery(BaseModel):
a: float
b: float
classSquareBody(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 --reload
4)有个很香的小功能:打开http://127.0.0.1:8000/docs你会看到自动生成的 Swagger UI,可以在页面里点点按钮、填个参数就调接口,非常适合甩给前端 or 测试同学。
同样地,前端或者其他服务,只要会发 HTTP 请求就能用你的逻辑了。
把“已有项目”的逻辑,挪到 API 里
上面的例子是从“函数”到“API”。 真实项目常见的情况是:你已经有一大堆业务逻辑,可能还封在类里、模块里,改起来不想动太多代码。
一个建议是:
不要把所有逻辑都写在 API 文件 API 只是“薄薄一层”:收参数 → 调用已有逻辑 → 返回结果
比如你原来有个 service.py:
# service.py
classPriceService:
def__init__(self, tax_rate: float = 0.1):
self.tax_rate = tax_rate
defcalc_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) # 比如税率写死或从配置加载
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}
这样做的好处是:
业务逻辑和接口层分开,单元测试好写 以后要换成 gRPC、命令行、定时任务调用,都可以复用 service.pyAPI 层只关心“收参数 + 做校验 + 错误码”
让 API 真的能抗住:简单说几句部署
本地 python app.py 开发没问题,但要给别人用、特别是要上服务器,还得注意两点:
1)用专门的服务启动工具
Flask:可以用 gunicorn + gevent 等 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
2)考虑 Docker 打包(这里只给个最简模板,有需要你再往下挖)
# 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
EXPOSE8000
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
这样你的 Python 代码就完全变成一个“黑盒服务”,谁都可以通过 HTTP 调用。
做 API 时几个比较容易忽略的小点
不整一堆 checklist,就挑几个实战里经常踩的:
1)输入一定要校验
别相信客户端传的永远是好数据。
用 FastAPI 就好办,参数类型写清楚,Pydantic 会帮你挡掉不少脏数据 自己手写的话,最起码 try/except + 判断必填字段
比如上面 Flask 的那段,其实已经是“最基础的校验”了。
2)错误别都返回 200
常见两个极端:
所有错误都返回 200+"success": false或者所有错误都直接 500,前端不知道发生了啥
比较自然的做法是:
参数问题 → 400 未登录/没权限 → 401/403 业务上找不到 → 404(或者 200 + 业务码,看你们团队约定) 真的内部炸了 → 500
FastAPI 里用 HTTPException 就挺顺手:
from fastapi import HTTPException
if something_wrong:
raise HTTPException(status_code=400, detail="参数不合法")
3)长耗时逻辑别直接让接口卡死
比如 slow_square 里我只是 sleep(1),可现实里可能是:
模型预测要几秒 调第三方接口很慢 做大文件处理
最简单的方式就是把“真正耗时的逻辑”丢到后台任务 / 消息队列里,这个就超出“快速变 API”的范围了,这里点一下就好。 先有最简单的同步版跑起来,把接口抽象清楚,再考虑异步、队列之类的优化。
一个迷你“从脚本到 API”的完整示例
最后拼一个非常完整又尽量短的小例子,你可以直接复制成一个目录跑起来:
目录结构:
my_api_demo/
calc.py
service.py
app.py
requirements.txt
calc.py:
defadd(a: float, b: float) -> float:
return a + b
service.py:
from calc import add
classCalcService:
defadd_with_log(self, a: float, b: float) -> float:
# 以后这里可以加日志、埋点、限流等等
result = add(a, b)
return result
app.py(用 FastAPI):
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from service import CalcService
app = 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:
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 函数 → 包一层 service → 暴露为 HTTP API → 本地/容器里都能跑”。
后面你想再加鉴权、限流、日志、监控、灰度发布之类,就在这个基础上慢慢往上盖楼就行了。
-END-
我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html
虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB