1. 为什么是FastAPI：爬虫工程师做接口时的真实痛点
先聊个我自己的经历。之前接了个需求，要把某个公开站点上的数据定时抓下来，整理成标准格式给下游系统调用。一开始的方案非常朴素：爬虫跑完往CSV里写，下游自己去读文件。结果没用两周就出问题了——下游改了字段格式，我这边清洗逻辑跟着改；他们想要增量数据，我CSV只能全量重写；更别提并发读取时文件锁冲突，整个链路经常红灯。
后来我意识到，这件事的正解不是“把文件做得更好”，而是“把数据变成接口”。于是选了FastAPI。
为什么是FastAPI而不是Flask或Django？说白了，爬虫工程里最烦的不是“能跑”，而是“数据格式不稳定”。今天页面加个字段，明天某个节点解析出来是None，后天接口突然返回一个空列表。这种情况下，FastAPI + Pydantic的组合几乎是为爬虫服务化量身定做的：
- 声明式响应模型让“接口返回什么”一目了然；
- 类型校验在请求入口就把脏数据挡在外面；
- 自带OpenAPI文档，下游对接的人不用追着你问字段含义；
- 原生异步支持，爬虫脚本里那些IO密集的活儿（请求、解析、写库）可以用async并发处理。
做个简单的横向对比，都是Python后端方案：
| 方案 | 类型校验 | 异步支持 | 自动API文档 | 适合场景 |
|---|---|---|---|---|
| Flask | 无（需手动） | 需额外插件 | 无 | 快速小Demo，内部工具 |
| Django + DRF | 有（Serializer） | 较弱 | 有 | 重业务、有后台管理需求 |
| FastAPI | 强（Pydantic） | 原生 | 强（Swagger UI） | 数据服务、爬虫接口、微服务 |
所以这篇文章不会去重复那些“Hello World”教程，而是把完整链路串一遍：从爬虫工作原理出发，到FastAPI接口设计，再到部署上线，最后附带面试里真正会问到的题目。整个过程都是我个人项目里跑过的方案，不是从文档里抄出来的理论。
2. 先理解爬虫的工作原理，再决定接口怎么设计
2.1 爬虫的四步核心流程
爬虫说穿了就是四件事：收集URL、发起请求、解析内容、存储与调度。看起来简单，但每一步都会影响你API的设计决策。
第一步，收集URL。可能是手工维护一批入口链接，也可能是从站点地图、翻页URL规则里自动生成。这一步决定了你的数据源有哪些“入口”。
第二步，发起请求。常见方案用requests或httpx，模拟浏览器请求头、处理Cookie、控制频率。这里有个关键点：爬虫是IO密集型任务，等待响应的时间占了绝大多数。所以爬虫服务化之后，接口层如果能用异步并发，性能和体验会差很多。
第三步，解析内容。BeautifulSoup处理HTML，正则抽字段，或者用json解析内嵌数据。这一步是“脏活”集中地——页面改版、字段缺失、编码异常，全在这一步爆发。
第四步，存储与调度。数据入库（SQLite/MySQL/MongoDB），定时触发增量更新，失败自动重试。
2.2 爬虫流程对API设计的三个约束
理解了这个流程，你会发现“给爬虫数据做API”不是一个单纯的CRUD工程，它有三个隐藏约束：
约束一：数据更新有延迟，接口要做“缓存层”。爬虫不可能实时抓取，更不可能用户每次请求都现抓一次——那样站点压力大，接口响应也慢得没法用。所以API层必须设计缓存或定时预取策略，让接口读的是快照数据，而不是实时爬取结果。
约束二：数据结构不稳定，接口要做“兜底策略”。页面字段可能缺失、类型可能变化。API如果直接把这些脏数据抛给下游，上游一小改，下游全崩。正确的做法是在接口层用Pydantic做一次严格校验和默认值兜底，宁可返回null，也不要让调用方收到非预期类型。
约束三：爬虫有频率和风控限制，接口要做“限流和鉴权”。如果把没鉴权的API暴露出去，被滥用后你的服务器IP可能被目标站点封掉，爬虫瞬间失效。所以接口上线前，鉴权和限流的优先级高于功能本身。
3. FastAPI项目结构怎么组织：从main.py单文件到分层目录
3.1 单文件为什么不够用
很多FastAPI教程喜欢把所有代码写进一个main.py，路由、模型、爬虫逻辑全放一起。说实话，只写20行Demo问题不大，但一旦出现“爬虫+多个数据源+定时任务+多个接口”，单文件代码就是灾难：
- 改一个字段要翻几百行；
- 新增数据源要改动已有路由；
- 测试根本无从下手；
- 部署时环境配置、日志配置全都耦合在一起。
3.2 我实际使用的目录结构
下面这个结构是我跑过几个项目之后沉淀下来的，不算复杂，但每个目录职责非常清晰：
crawler_api/ ├── main.py # FastAPI入口，创建app、挂载路由 ├── core/ │ ├── config.py # 读取环境变量、配置项 │ ├── logging.py # 统一日志配置 │ └── security.py # API Key校验、限流依赖 ├── models/ │ ├── schemas.py # Pydantic请求/响应模型 │ └── database.py # 数据库连接 ├── api/ │ ├── v1/ │ │ ├── endpoints/ │ │ │ ├── items.py │ │ │ └── rank.py │ │ └── router.py # 汇总路由 ├── services/ │ └── item_service.py # 业务逻辑：查询、缓存、聚合 ├── crawlers/ │ ├── base.py # 爬虫基类，封装请求重试 │ ├── source_a.py # 数据源A的具体爬虫 │ └── scheduler.py # 定时采集任务 └── requirements.txt这个结构的关键在于分层：
- api层只做参数接收和响应返回，不写业务逻辑；
- services层负责数据处理，比如查缓存、查数据库、组装聚合数据；
- crawlers层独立出来，不依赖FastAPI概念，换到别的框架也能复用。
3.3 main.py里应该有什么
main.py只做三件事：创建FastAPI实例、加载配置、注册路由。举一个精简但完整的例子：
from fastapi import FastAPI from core.config import settings from api.v1.router import api_router app = FastAPI( title=settings.PROJECT_NAME, openapi_url=f"{settings.API_PREFIX}/openapi.json", ) app.include_router(api_router, prefix=settings.API_PREFIX) @app.on_event("startup") async def startup_event(): from crawlers.scheduler import start_scheduler start_scheduler()这里有个细节：openapi_url我习惯带上前缀，避免和后面Nginx的路由冲突。
4. 核心代码实现：爬虫采集到接口响应的完整链路
4.1 爬虫模块的基类设计
写爬虫最容易犯的错是“每个数据源写一套完全不同的逻辑”。我建议先做一个基类，把公共的请求、重试、UA伪装、超时控制都封装好，然后再针对单个数据源做子类。
import time import random import requests from abc import ABC, abstractmethod class BaseCrawler(ABC): BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } def __init__(self, timeout=10, retries=3, delay_range=(1, 3)): self.timeout = timeout self.retries = retries self.delay_range = delay_range def _safe_request(self, url, **kwargs): for attempt in range(self.retries): try: resp = requests.get(url, headers=self.BASE_HEADERS, timeout=self.timeout, **kwargs) resp.raise_for_status() return resp except requests.RequestException as e: if attempt == self.retries - 1: logger.error(f"请求失败: {url}, error={e}") return None time.sleep(2 ** attempt) # 指数退避 def _polite_sleep(self): time.sleep(random.uniform(*self.delay_range)) @abstractmethod def fetch_data(self): """子类实现具体数据源的抓取与解析""" pass最值得说的是_polite_sleep和指数退避。前者是在两次请求之间随机休眠，降低被封的风险；后者是请求失败后按2的幂次等待再重试，避免刚失败就立刻猛冲把站点打爆。这些细节不是“为了优雅”，而是爬虫服务能不能长期稳定跑下去的关键。
4.2 Pydantic模型：接口的“守门员”
Pydantic模型承担了三个职责：校验请求参数、定义响应结构、给不稳定字段设置默认值。
from pydantic import BaseModel, Field from typing import Optional from datetime import datetime class ItemOut(BaseModel): id: int title: str price: float = Field(default=0.0, ge=0) source_url: str created_at: Optional[datetime] = None这里price字段加了默认值和ge=0校验。为什么？因为爬虫拿到的价格字段可能为空字符串，可能带单位，甚至可能是"面议"。如果直接转float会抛异常，但交给Pydantic后，解析失败就用默认值0.0，接口不会因为一条脏数据整个崩溃。
4.3 路由层：保持简洁
from fastapi import APIRouter, Depends, Query from models.schemas import ItemOut from services.item_service import get_items_with_cache from api.v1.deps import verify_api_key router = APIRouter(prefix="/items", tags=["items"]) @router.get("", response_model=list[ItemOut]) def list_items( keyword: str = Query(None, max_length=50), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), _: str = Depends(verify_api_key), ): data = get_items_with_cache(keyword=keyword, page=page, page_size=page_size) return data注意两点：一是response_model显式声明，FastAPI会自动过滤掉模型里没有的字段。二是依赖注入Depends(verify_api_key)统一做鉴权，路由函数本身不需要关心API Key逻辑。
4.4 Services层：缓存策略
爬虫数据接口最常见的缓存方案是TTL缓存。我常用cachetools或干脆用Redis，取决于部署规模。这里展示一个基于内存TTL的简单实现：
from cachetools import TTLCache from datetime import datetime cache = TTLCache(maxsize=128, ttl=300) # 缓存5分钟 def get_items_with_cache(keyword=None, page=1, page_size=20): cache_key = f"items:{keyword}:{page}:{page_size}" if cache_key in cache: return cache[cache_key] # 这里原本是从数据库查询数据 data = query_from_database(keyword=keyword, offset=(page-1)*page_size, limit=page_size) # 记录缓存时间，方便排查问题 data = {"items": data, "cached_at": datetime.now().isoformat()} cache[cache_key] = data return dataTTL缓存最大的好处是扛住瞬时并发。哪怕爬虫每半小时才更新一次数据，用户狂点时接口也能在毫秒级响应，而不会5分钟触发一次大规模爬取。
4.5 定时采集任务
FastAPI没有内置定时任务，但可以把apscheduler挂进生命周期里。调度器负责每30分钟启动一次爬虫，把数据写入数据库并清空相关缓存。
from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() def start_scheduler(): scheduler.add_job(run_crawl_all, "interval", minutes=30, id="crawl_all") scheduler.start()这里最大的坑是：爬虫任务不要和API进程共用日志文件，也不要共用内存缓存。因为定时任务一旦OOM或崩溃，可能会拖垮整个API进程。有条件的话，把采集器和API服务拆成两个独立进程，这样采集器挂了API还能返回旧数据。
5. 接口上线前必须做的：鉴权、限流、CORS和异常兜底
5.1 鉴权：API Key还是JWT？
爬虫接口的使用方往往是内部系统或少数外部合作方，我倾向于用简单的API Key，而不是JWT。原因很实在：
- JWT有过期时间，调用方需要维护刷新逻辑，提高了对接成本；
- API Key就是一个随机字符串，放在请求头里，服务端比对即可；
- 如果是内部服务，还可以结合IP白名单双保险。
生成API Key可以用secrets.token_urlsafe(32)，存数据库时只存哈希，别存明文。实际项目里我遇到过有人把Key硬编码在代码里然后提交到Git仓库的，这种低级错误真的会连累整个服务器被封。
5.2 限流：没有限流的爬虫API是定时炸弹
限流逻辑实现方式很多，我试过用slowapi，也手写过简单的内存限流。如果不想引入太多依赖，可以基于itsdangerous写一个简单的滑动窗口限流。但如果项目规模到了生产级，我建议直接用Redis处理限流计数，因为单机内存方案在多进程部署下会失效。
这里必须强调一个面试高频点：为什么不能用Per-Process的内存计数做限流。因为生产环境通常用Gunicorn开多个worker进程，每个进程的内存是独立的，一个IP在A进程请求了10次，再去B进程还能请求10次，限流直接失效。Redis方案是原子的，才能保证全局限流。
5.3 异常兜底与统一响应格式
爬虫接口最大的特点就是“上游数据不稳定”。所以全局异常处理必须覆盖两类情况：
一是爬虫数据解析异常，比如字段缺失、类型转换失败。这类要捕获后返回默认值，不能把异常堆栈直接吐给调用方。
二是第三方调用方传参错误。FastAPI的RequestValidationError默认返回422格式，但你不希望下游看到一串专业术语。建议重写异常处理方法，返回一个固定的错误结构：
from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): return JSONResponse( status_code=400, content={ "code": "INVALID_PARAM", "detail": "请求参数不合法", "errors": exc.errors(), }, )统一错误格式对下游非常友好。联调时可以直接告诉对方：“只要看到code字段就不是成功”，比他们自己解析不同风格的报错要省心得多。
5.4 CORS设置
如果接口要供Web前端调用，别忘了配置CORS。很多爬虫接口是给别人后端调用的，CORS反而可能不需要。但如果要开放给浏览器端调试，不配置的话对方前端控制台会直接报跨域错误。
6. 部署方案：Docker + Uvicorn + Nginx一套走通
6.1 Docker镜像怎么写
写FastAPI的Dockerfile时有几个细节值得注意：
FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]第一，不要用python:3.11这种带完整工具链的镜像，slim就够了，能少几百MB。第二，--workers 4要结合CPU核数设置，不是越大越好。第三，--host 0.0.0.0必须写，否则容器外部访问不到。
6.2 docker-compose编排爬虫服务
如果采集和API拆成两个服务，compose文件大概长这样：
version: "3.8" services: api: build: . ports: - "8000:8000" env_file: - .env depends_on: - redis crawler: build: context: . dockerfile: Dockerfile.crawler env_file: - .env depends_on: - redis这里crawler是一个独立镜像，启动后直接跑scheduler.py，不暴露任何端口。好处前面说过——采集挂了不影响API服务返回旧缓存数据。
6.3 Nginx反向代理与健康检查
Nginx层要做的事有两件：把/api前缀转发到FastAPI，以及配置健康检查。简单配置如下：
location /api/ { proxy_pass http://127.0.0.1:8000/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; }关于proxy_pass末尾的/，这里有个经典坑：如果写proxy_pass http://127.0.0.1:8000;（不带斜杠），请求/api/items会原样转发给后端；如果带斜杠，则/api/会被替换成/，后端收到的是/items。两种都对，但必须和FastAPI路由前缀保持一致，否则404查半天都不知道问题在哪。
6.4 部署时最容易忽略的环境问题
我踩过最大的坑是时区。容器默认UTC时间，爬虫定时任务如果按本地时间调度，会出现“每天凌晨3点跑”和“早上9点跑”这种乌龙。解决办法是在Docker启动参数里加TZ=Asia/Shanghai，或者在代码里统一用datetime.now(timezone.utc)后再换算。
另一个坑是健康检查接口。部署平台（比如K8s）会定期请求/health，如果你没有这个接口，服务会被判定不健康。加一个就行：
@app.get("/health") def health_check(): return {"status": "ok"}7. 面试习题解析：这些题目到底在考什么
标题里提到了面试习题，我根据自己的面经把和FastAPI、爬虫服务化相关的题目做个分类拆解。这些题是我和不同的候选人聊过的，也踩过不少“答不到点子上”的情况。
7.1 问FastAPI：注意区分“背概念”和“真会用”
Q1：FastAPI为什么性能比Flask好？
面试官真正想听的，不是“因为它快”，而是异步和非阻塞IO。FastAPI基于Starlette，天然支持asyncio，在处理IO密集型任务（查数据库、调外部API）时可以用async def并发执行。Flask是WSGI同步模型，一个请求占用一个线程，高并发时线程切换开销大。但注意：如果你把async def当普通def写同步阻塞代码，性能反而更差。这一点要主动说，面试官会觉得你真的踩过坑。
Q2：Pydantic的validator和field_validator怎么用？
高频坑题。新版Pydantic v2用@field_validator，旧版是@validator。很多人项目升级后代码直接报错。更深一层，面试官想考的是“你知不知道响应模型里的校验逻辑是服务端最后一道防线”。比如爬一个价格字段时，页面可能返回"价格面议"，就可以在validator里转换：
from pydantic import field_validator class ItemOut(BaseModel): price: float = 0.0 @field_validator("price", mode="before") @classmethod def parse_price(cls, v): if isinstance(v, str): v = v.replace("元", "").replace(",", "") try: return float(v) except ValueError: return 0.0 return vQ3：FastAPI如何做依赖注入？
依赖注入本质是Depends()，它的优势是“声明式”。面试官会顺带问：“依赖里能不能用async？”——可以，但要注意普通def依赖是线程池执行、async def依赖是事件循环执行，两者在事务管理上有细微差别。
7.2 问爬虫：不是问你会不会解析，而是链路设计
Q4：爬虫被封了怎么办？
标准回答链路是：降低频率、随机UA、代理池、分布式采集、遵守robots协议。但面试官更想听的可能是“根据被封的维度做针对性处理”——比如封IP就走代理池，封账号就做Cookie池，封特征就做浏览器指纹伪装。重点是要有条理，而不是堆名词。
Q5：如何设计一个爬虫数据服务，保证下游能稳定调用？
这题几乎就是本文的核心场景。回答时可以按“采集层-存储层-接口层”三段式展开：
- 采集层：定时增量更新，失败重试，反爬策略；
- 存储层：上数据库，做好去重和索引，保留数据历史版本；
- 接口层：Pydantic统一格式、缓存兜住并发、鉴权和限流保证服务安全。
能按这个逻辑答出来，基本就是一个合格的爬虫后端工程师思路。
Q6：爬虫拿到的数据有脏数据、有缺失，接口怎么处理？
答案是Pydantic默认值+前端接口容错。从技术层面保证接口返回的结构永远是“合法”的，把脏数据清洗放到采集阶段做，接口层只做格式强制校验。这和本文4.2的代码逻辑一致。
7.3 问部署：这些坑不实际跑过很难答好
Q7：Gunicorn worker数和Uvicorn worker数什么关系？
最烦这种概念混淆题。简单结论是：Gunicorn管理worker进程，Uvicorn是ASGI服务器，可以直接用gunicorn -k uvicorn.workers.UvicornWorker让Gunicorn启动Uvicorn worker。worker数通常设为CPU核心数的2~4倍，但不是线性增长的——过多worker会导致上下文切换开销大于收益，还可能打满数据库连接池。
Q8：API服务内存不断上涨怎么办？
抛开代码泄露问题，HTTP客户端连接泄漏和缓存无限增长是最常见的两个原因。比如requests未关闭会话、TTLCache没有指定maxsize。还有一个隐蔽点：如果用了异步爬虫库，协程没有正确关闭，也会堆积内存。这类题最重要是表现出“排查链路”能力：先从监控看是哪个进程的内存涨，再用tracemalloc或memory_profiler定位是哪个数据结构的增长。
8. 进阶优化：从“能用”到“好用的服务”还要做什么
8.1 数据库选型与表结构设计
爬虫数据服务我推荐先用SQLite或PostgreSQL。表结构上要特别注意“更新时间”和“唯一键”的设计：
- 唯一键：数据源URL天然是唯一键，用于去重和增量更新；
- 更新时间：要有
updated_at索引，否则回源查询慢； - 保留原始字段：在清洗后的字段旁保留一份原始JSON，方便排查数据源变更。
8.2 监控与告警：接口挂了要第一时间知道
生产环境一定要有监控。最简单的方式是部署到云服务后配置一个HTTP探活，每隔30秒请求一次/health和核心数据接口。如果接口超过5秒没响应就告警。
8.3 接口版本化
接口一上线就不可能随便改字段名，所以API要带版本号。我用的是/api/v1前缀，同时把v2的代码放在api/v2目录下，两个版本可以共存。爬虫数据源变化很频繁，版本化能让你在数据格式大改时不用强迫下游立刻升级。
9. 写在最后：爬虫服务化这件事的“底线”
通篇讲了很多技术细节，最后说点实际的。爬虫和接口开发本身没有原罪，但在做任何数据采集前，我都建议先确认三件事：目标站点是否允许爬取（robots.txt）、你采集的数据是否涉及个人隐私、以及你的采集频率会不会影响对方正常服务。
我接手这类项目时，会优先评估“数据获取是否合规”这层，而不是先写代码。接口做得再好、再稳定，如果数据来源本身有问题，对整个项目都是定时炸弹。
另外，从我实际开发的经验来看，最好的路径应该是“小步快跑”：先用一段脚本验证数据源能抓、能解析，再花时间设计项目结构，最后才是接口上线和部署优化。不要一上来就把目录分得特别复杂，因为你对数据源的理解往往会在写代码的过程中刷新好几次。先跑通，再重构，比一开始就设计一个“完美架构”要靠谱得多。
