FastAPI入门:Python后端开发新手指南 —— 一个百度搜索算法工程师的“被迫转型”实录
去年双11前夜,我还在百度搜索部门调模型参数,盯着线上CTR指标掉得比我的发际线还快。产品经理凌晨三点在钉钉群里@我:“这个新需求能不能用接口暴露出去?前端要接。”
我回了个“?”,心想:我又不是后端,我是搞搜索排序的算法工程师啊!
但现实很骨感——在百度这种大厂,“全栈能力”早已不是加分项,而是保命符。尤其最近公司内部推“微服务化”,连我们算法组都要自己搭接口、自测、上线、扛流量。更别提我这三年多待得有点腻了,简历上除了“熟悉XGBoost”和“调过BERT”,得有点能打动新东家的东西。
于是,FastAPI 出现在了我的视野里。
为啥是 FastAPI?而不是 Flask 或 Django?
说实话,一开始我也想直接上 Flask——毕竟大学写课设就用它,简单、轻量、文档友好。但当我看到团队里新人用 FastAPI 三天搭出一个带自动文档、类型校验、异步支持的接口服务时,我酸了。
更重要的是,FastAPI 的设计理念和现代 Python 开发范式高度契合:
- 基于 Pydantic 的数据校验(告别
if not isinstance(x, str):的祖传代码) - 自动生成 OpenAPI 文档(再也不用求着前端看 Swagger)
- 原生支持 async/await(对高并发场景友好)
- 性能接近 Node.js(官方 benchmark 比 Flask 快好几倍)
最关键的一点:它让我这个“半吊子后端”也能写出看起来很专业的 API。
实战案例:搭建一个区块链地址查询服务
上周五晚上,我被领导临时抓壮丁:“有个小项目,对接第三方区块链浏览器 API,提供本地缓存+查询接口,下周三上线。”
我:???这不就是 CRUD 吗?但加上“区块链”三个字,听起来立马高大上了(简历又能加一行)。
需求拆解
- 用户传入一个 Ethereum 地址(如
0xAb58...) - 后端调用 Etherscan API 获取余额、交易数等信息
- 结果缓存到 Redis(避免频繁请求第三方)
- 提供
/address/{addr}接口返回结构化 JSON - 自动校验地址格式(必须以
0x开头 + 40 位十六进制)
听起来简单?但在我第一次部署到测试环境时,运维小哥直接甩来一句:“你这接口没做限流,压测直接打爆了 Etherscan 的 rate limit,人家 IP 封了我们两小时。”
当时真的想砸电脑。
项目搭建:从零开始 FastAPI
1. 环境初始化
mkdir eth-query-service && cd eth-query-service
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
pip install fastapi uvicorn httpx redis pydantic python-dotenv
插件党福利:我在 VSCode 里装了 Pylance、Black Formatter、Python Docstring Generator,写代码自动补全+格式化一条龙,效率拉满。
2. 核心代码:main.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, validator
import httpx
import redis.asyncio as redis
import os
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="ETH Address Query Service", version="0.1.0")
# Redis 连接池
redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
class EthAddress(BaseModel):
address: str
@validator("address")
def validate_eth_address(cls, v):
if not v.startswith("0x") or len(v) != 42:
raise ValueError("Invalid Ethereum address format")
if not all(c in "0123456789abcdefABCDEF" for c in v[2:]):
raise ValueError("Address contains invalid hex characters")
return v.lower() # 统一小写
async def fetch_from_etherscan(addr: str):
api_key = os.getenv("ETHERSCAN_API_KEY")
url = f"https://api.etherscan.io/api?module=account&action=balance&address={addr}&tag=latest&apikey={api_key}"
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.get(url)
data = resp.json()
if data["status"] != "1":
raise HTTPException(status_code=502, detail="Etherscan API error")
return {
"address": addr,
"balance_wei": int(data["result"]),
"balance_eth": int(data["result"]) / 1e18,
}
@app.get("/address/{addr}", summary="Get ETH balance by address")
async def get_address_info(addr: str):
# 先查缓存
cached = await redis_client.get(f"eth:{addr}")
if cached:
import json
return json.loads(cached)
# 校验地址(这里其实可以提前做,但为了演示放这里)
try:
validated = EthAddress(address=addr)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
# 调第三方 API
try:
result = await fetch_from_etherscan(validated.address)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Fetch failed: {str(e)}")
# 写入缓存,TTL 5分钟
await redis_client.setex(f"eth:{addr}", 300, str(result).replace("'", '"'))
return result
3. 配置文件:.env
REDIS_URL=redis://localhost:6379/0
ETHERSCAN_API_KEY=your_api_key_here
4. 启动服务
uvicorn main:app --reload --port 8000
访问 http://localhost:8000/docs,自动弹出 Swagger UI,接口文档、测试表单一键搞定。前端同事看到后直呼“这比我们老系统强100倍”。
踩坑实录:那些让我半夜惊醒的 Bug
❌ 坑1:Redis 缓存序列化问题
最开始我直接 redis.set(key, dict),结果读出来是 <class 'str'>,解析失败。后来发现 Redis 只存字符串,必须手动 JSON 序列化/反序列化。
改进方案:用
json.dumps(result)存,json.loads()取。或者更优雅地,封装一个缓存装饰器。
❌ 坑2:没做请求限流,被 Etherscan 拉黑
FastAPI 本身不带限流,得靠中间件。后来我加了 slowapi:
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/address/{addr}")
@limiter.limit("5/minute") # 每IP每分钟5次
async def get_address_info(addr: str, request: Request):
...
上线后再也没被封 IP。
❌ 坑3:异步 Redis 连接泄露
一开始用 redis.Redis()(同步版),在 async 函数里阻塞了事件循环,QPS 直接掉一半。换成 redis.asyncio 才解决。
生产环境 Checklist(来自百度血泪经验)
在百度,我们有一套严格的上线流程。FastAPI 虽然轻量,但生产环境不能马虎:
| 项目 | 是否完成 | 说明 |
|---|---|---|
| 日志记录 | ✅ | 用 logging 模块,关键路径打 INFO,错误打 ERROR |
| 健康检查 | ✅ | 加 /health 接口,K8s 用 |
| 错误监控 | ✅ | 接入 Sentry,捕获未处理异常 |
| Docker 化 | ✅ | 写 Dockerfile,避免“在我机器上能跑” |
| 环境隔离 | ✅ | dev/test/prod 三套 .env |
| 自动文档 | ✅ | FastAPI 自带,但记得关掉 prod 的 /docs |
Dockerfile 示例:
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
性能对比:FastAPI vs Flask(实测数据)
我在本地用 wrk 压测了两个几乎相同的接口(无数据库,纯 JSON 返回):
| 框架 | 并发 100 | QPS | 平均延迟 |
|---|---|---|---|
| Flask (sync) | 100 | ~850 | 118ms |
| FastAPI (async) | 100 | ~2400 | 42ms |
测试环境:MacBook Pro M1, Python 3.10
注意:真实场景中,如果涉及数据库 I/O,FastAPI 的 async 优势会更明显。
写在最后:为什么我推荐 FastAPI 给 Python 新手?
作为一个在百度写了三年 CTR 模型、对 Web 开发一度恐惧的算法工程师,FastAPI 让我重新认识了“后端开发”——它不是玄学,而是一套可验证、可测试、可自动化的工程实践。
而且,FastAPI 的类型提示 + Pydantic 模型,天然适合我们这些习惯写 def train(X: np.ndarray, y: np.ndarray) -> Model 的人。代码即文档,文档即契约,这不就是我们追求的“确定性”吗?
至于跳槽?简历上已经加上了:“主导设计并落地基于 FastAPI 的区块链数据查询服务,支持日均 50W+ 请求,P99 延迟 < 200ms”。虽然实际就我一个人写……但听起来是不是很厉害?
如果你也在大厂被逼着“既要会算法又要会接口”,或者想用 Python 快速搭个 MVP,别犹豫,上 FastAPI。它可能不会让你成为后端大神,但至少能让你在周五下班前,不用再被产品经理钉钉轰炸。
附:完整代码已上传 GitHub(私信我拿链接,防止被说打广告)。
对了,如果你也在考虑换环境,欢迎交流~ 我的目标:下一份工作,键盘不沾泡面渣。

评论 0