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( titlesettings.PROJECT_NAME, openapi_urlf{settings.API_PREFIX}/openapi.json, ) app.include_router(api_router, prefixsettings.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, timeout10, retries3, 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, headersself.BASE_HEADERS, timeoutself.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(default0.0, ge0) source_url: str created_at: Optional[datetime] None这里price字段加了默认值和ge0校验。为什么因为爬虫拿到的价格字段可能为空字符串可能带单位甚至可能是面议。如果直接转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_modellist[ItemOut]) def list_items( keyword: str Query(None, max_length50), page: int Query(1, ge1), page_size: int Query(20, ge1, le100), _: str Depends(verify_api_key), ): data get_items_with_cache(keywordkeyword, pagepage, page_sizepage_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(maxsize128, ttl300) # 缓存5分钟 def get_items_with_cache(keywordNone, page1, page_size20): cache_key fitems:{keyword}:{page}:{page_size} if cache_key in cache: return cache[cache_key] # 这里原本是从数据库查询数据 data query_from_database(keywordkeyword, offset(page-1)*page_size, limitpage_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, minutes30, idcrawl_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_code400, 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启动参数里加TZAsia/Shanghai或者在代码里统一用datetime.now(timezone.utc)后再换算。另一个坑是健康检查接口。部署平台比如K8s会定期请求/health如果你没有这个接口服务会被判定不健康。加一个就行app.get(/health) def health_check(): return {status: ok}7. 面试习题解析这些题目到底在考什么标题里提到了面试习题我根据自己的面经把和FastAPI、爬虫服务化相关的题目做个分类拆解。这些题是我和不同的候选人聊过的也踩过不少“答不到点子上”的情况。7.1 问FastAPI注意区分“背概念”和“真会用”Q1FastAPI为什么性能比Flask好面试官真正想听的不是“因为它快”而是异步和非阻塞IO。FastAPI基于Starlette天然支持asyncio在处理IO密集型任务查数据库、调外部API时可以用async def并发执行。Flask是WSGI同步模型一个请求占用一个线程高并发时线程切换开销大。但注意如果你把async def当普通def写同步阻塞代码性能反而更差。这一点要主动说面试官会觉得你真的踩过坑。Q2Pydantic的validator和field_validator怎么用高频坑题。新版Pydantic v2用field_validator旧版是validator。很多人项目升级后代码直接报错。更深一层面试官想考的是“你知不知道响应模型里的校验逻辑是服务端最后一道防线”。比如爬一个价格字段时页面可能返回价格面议就可以在validator里转换from pydantic import field_validator class ItemOut(BaseModel): price: float 0.0 field_validator(price, modebefore) classmethod def parse_price(cls, v): if isinstance(v, str): v v.replace(元, ).replace(,, ) try: return float(v) except ValueError: return 0.0 return vQ3FastAPI如何做依赖注入依赖注入本质是Depends()它的优势是“声明式”。面试官会顺带问“依赖里能不能用async”——可以但要注意普通def依赖是线程池执行、async def依赖是事件循环执行两者在事务管理上有细微差别。7.2 问爬虫不是问你会不会解析而是链路设计Q4爬虫被封了怎么办标准回答链路是降低频率、随机UA、代理池、分布式采集、遵守robots协议。但面试官更想听的可能是“根据被封的维度做针对性处理”——比如封IP就走代理池封账号就做Cookie池封特征就做浏览器指纹伪装。重点是要有条理而不是堆名词。Q5如何设计一个爬虫数据服务保证下游能稳定调用这题几乎就是本文的核心场景。回答时可以按“采集层-存储层-接口层”三段式展开采集层定时增量更新失败重试反爬策略存储层上数据库做好去重和索引保留数据历史版本接口层Pydantic统一格式、缓存兜住并发、鉴权和限流保证服务安全。能按这个逻辑答出来基本就是一个合格的爬虫后端工程师思路。Q6爬虫拿到的数据有脏数据、有缺失接口怎么处理答案是Pydantic默认值前端接口容错。从技术层面保证接口返回的结构永远是“合法”的把脏数据清洗放到采集阶段做接口层只做格式强制校验。这和本文4.2的代码逻辑一致。7.3 问部署这些坑不实际跑过很难答好Q7Gunicorn worker数和Uvicorn worker数什么关系最烦这种概念混淆题。简单结论是Gunicorn管理worker进程Uvicorn是ASGI服务器可以直接用gunicorn -k uvicorn.workers.UvicornWorker让Gunicorn启动Uvicorn worker。worker数通常设为CPU核心数的2~4倍但不是线性增长的——过多worker会导致上下文切换开销大于收益还可能打满数据库连接池。Q8API服务内存不断上涨怎么办抛开代码泄露问题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、你采集的数据是否涉及个人隐私、以及你的采集频率会不会影响对方正常服务。我接手这类项目时会优先评估“数据获取是否合规”这层而不是先写代码。接口做得再好、再稳定如果数据来源本身有问题对整个项目都是定时炸弹。另外从我实际开发的经验来看最好的路径应该是“小步快跑”先用一段脚本验证数据源能抓、能解析再花时间设计项目结构最后才是接口上线和部署优化。不要一上来就把目录分得特别复杂因为你对数据源的理解往往会在写代码的过程中刷新好几次。先跑通再重构比一开始就设计一个“完美架构”要靠谱得多。