FastAPI与异步爬虫的完美融合：构建高性能房源搜索API
在之前的文章中，我们实现了基于asyncio的多源竞速爬虫核心逻辑。本文将重点讲解如何将这套爬虫系统与FastAPI框架深度融合，打造一个生产级的房源搜索API服务。
一、 为什么选择FastAPI？
FastAPI 作为当前最热门的Python Web框架之一，与异步爬虫有着天然的契合度：
- 原生异步支持：基于Starlette和Pydantic，天然支持
async/await，与asyncio爬虫无缝衔接 - 高性能：性能与Node.js和Go相当，远超Flask/Django
- 自动文档：自动生成Swagger UI和ReDoc文档，方便前后端联调
- 类型提示驱动：利用Python类型注解，实现请求/响应的自动校验和序列化
二、 路由层设计：轻量而清晰
路由模块的职责非常单一——接收HTTP请求，调用服务层，返回JSON响应。
from fastapi import APIRouter, Request from api.crawler.services.crawler_service import crawl_housing_data router = APIRouter() @router.get("/status") def get_crawler_status(): """获取爬虫服务状态""" return {"status": "running", "message": "爬虫服务正常"} @router.post("/search") async def search_housing_listings(request: Request, request_data: dict): """搜索房源列表""" return await crawl_housing_data(request, request_data)关键点解析
1. 同步 vs 异步路由
/status是一个简单的健康检查接口，不涉及IO操作，使用同步函数即可/search需要等待爬虫完成网络请求，必须声明为async def，这样FastAPI会在事件循环中执行它，不会阻塞其他请求
2.Request对象的作用
注意search_housing_listings的第一个参数是Request。我们传入它的核心目的是获取应用级别的共享资源：
browser_context = request.app.state.browser_contextrequest.app.state是FastAPI提供的全局状态存储，非常适合存放数据库连接池、浏览器上下文等跨请求复用的重型资源。
三、 服务层融合：从HTTP到爬虫
服务层是路由和爬虫核心之间的桥梁，它完成了三件事：
async def crawl_housing_data(request: Request, request_data: Dict[str, Optional[str]]): # 1. 参数转换：将用户输入转为各平台URL parsed_data = parse_keywords_to_url2(request_data) # 2. 获取共享资源并执行竞速爬虫 browser_context = request.app.state.browser_context data = await race_spiders(parsed_data, browser_context) # 3. 异常兜底，保证API永远返回合法JSON return data if data else []融合要点
- 参数适配：
parse_keywords_to_url2将前端传来的{city, district, price...}转换为{anjuke: {url, city...}, lianjia: {...}}的多平台URL字典 - 资源注入：通过
request.app.state获取预先初始化的浏览器上下文，避免每次请求都启动新浏览器 - 异常隔离：爬虫层的任何异常都不会穿透到HTTP层，保证API始终返回
200 OK和合法JSON
四、 应用生命周期管理：优雅地启动与关闭
这是FastAPI与爬虫融合中最关键的一环。浏览器（如Playwright）是重型资源，必须在应用启动时初始化，在关闭时释放。
from fastapi import FastAPI from playwright.async_api import async_playwright app = FastAPI() @app.on_event("startup") async def startup_event(): """应用启动时初始化浏览器上下文""" playwright = await async_playwright().start() browser = await playwright.chromium.launch(headless=True) context = await browser.new_context() # 挂载到app.state，全局共享 app.state.playwright = playwright app.state.browser = browser app.state.browser_context = context print("🚀 浏览器上下文初始化完成") @app.on_event("shutdown") async def shutdown_event(): """应用关闭时清理资源""" if hasattr(app.state, "browser_context"): await app.state.browser_context.close() if hasattr(app.state, "browser"): await app.state.browser.close() if hasattr(app.state, "playwright"): await app.state.playwright.stop() print("🛑 浏览器资源已释放")⚠️注意：FastAPI 0.99+ 推荐使用
lifespan替代@app.on_event，写法更现代：这里因为lifespan使用出现错误改为旧版@app.on_event事件
from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 playwright = await async_playwright().start() browser = await playwright.chromium.launch(headless=True) app.state.browser_context = await browser.new_context() yield # 关闭逻辑 await app.state.browser_context.close() await browser.close() await playwright.stop() app = FastAPI(lifespan=lifespan)五、 请求校验：用Pydantic提升接口质量
直接使用dict接收请求虽然灵活，但缺乏校验。在生产环境中，建议使用Pydantic模型：
from pydantic import BaseModel, Field from typing import Optional class HousingSearchRequest(BaseModel): city: str = Field(..., description="城市名称", example="济南") district: Optional[str] = Field(None, description="区域", example="历下区") rent_type: Optional[str] = Field(None, description="租赁类型", example="整租") rooms: Optional[int] = Field(None, description="房间数", example=3) price: Optional[str] = Field(None, description="价格上限", example="2000") orientation: Optional[str] = Field(None, description="朝向", example="南") @router.post("/search") async def search_housing_listings(request: Request, request_data: HousingSearchRequest): # Pydantic自动完成类型转换和校验 return await crawl_housing_data(request, request_data.dict())这样做的好处：
- 自动返回422错误和清晰的校验信息
- 自动生成OpenAPI文档中的请求体示例
- 类型安全，IDE自动补全
六、 完整项目结构
api/ ├── crawler/ │ ├── __init__.py │ ├── main_runner.py # 竞速引擎 │ ├── services/ │ │ └── crawler_service.py # 服务层 │ ├── spiders/ │ │ ├── AnjukeSpider.py │ │ ├── FangSpider.py │ │ └── LianjiaSpiderAsync.py │ └── util/ │ └── parseToUrl.py ├── routes/ │ └── crawler_router.py # 路由层 └── main.py # FastAPI入口main.py入口文件：
from fastapi import FastAPI from api.routes.crawler_router import router as crawler_router app = FastAPI(title="房源爬虫API", version="1.0.0") # 注册路由 app.include_router(crawler_router, prefix="/api/crawler", tags=["爬虫服务"]) @app.get("/") def root(): return {"message": "房源爬虫服务已启动"}七、 测试与验证
启动服务后，访问以下地址：
- 健康检查：
GET http://localhost:8000/api/crawler/status - API文档：
http://localhost:8000/docs（Swagger UI） - 搜索接口：
POST http://localhost:8000/api/crawler/search
