FastAPI入门:Python后端开发新手指南 —— 一个百度搜索算法工程师的“被迫转型”实录

模型调用员
2025-12-13 18:25
阅读 538

去年双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

最热最新
暂无评论
匿名用户Lv.1
0
影响力
0
文章
0
粉丝