FastAPI与异步爬虫的完美融合构建高性能房源搜索API在之前的文章中我们实现了基于asyncio的多源竞速爬虫核心逻辑。本文将重点讲解如何将这套爬虫系统与FastAPI框架深度融合打造一个生产级的房源搜索API服务。一、 为什么选择FastAPIFastAPI 作为当前最热门的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(headlessTrue) 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(headlessTrue) app.state.browser_context await browser.new_context() yield # 关闭逻辑 await app.state.browser_context.close() await browser.close() await playwright.stop() app FastAPI(lifespanlifespan)五、 请求校验用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房间数, example3) price: Optional[str] Field(None, description价格上限, example2000) 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, version1.0.0) # 注册路由 app.include_router(crawler_router, prefix/api/crawler, tags[爬虫服务]) app.get(/) def root(): return {message: 房源爬虫服务已启动}七、 测试与验证启动服务后访问以下地址健康检查GET http://localhost:8000/api/crawler/statusAPI文档http://localhost:8000/docsSwagger UI搜索接口POST http://localhost:8000/api/crawler/search
项目实训——大数据租房推荐智能体（爬虫部分7）
FastAPI与异步爬虫的完美融合构建高性能房源搜索API在之前的文章中我们实现了基于asyncio的多源竞速爬虫核心逻辑。本文将重点讲解如何将这套爬虫系统与FastAPI框架深度融合打造一个生产级的房源搜索API服务。一、 为什么选择FastAPIFastAPI 作为当前最热门的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(headlessTrue) 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(headlessTrue) app.state.browser_context await browser.new_context() yield # 关闭逻辑 await app.state.browser_context.close() await browser.close() await playwright.stop() app FastAPI(lifespanlifespan)五、 请求校验用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房间数, example3) price: Optional[str] Field(None, description价格上限, example2000) 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, version1.0.0) # 注册路由 app.include_router(crawler_router, prefix/api/crawler, tags[爬虫服务]) app.get(/) def root(): return {message: 房源爬虫服务已启动}七、 测试与验证启动服务后访问以下地址健康检查GET http://localhost:8000/api/crawler/statusAPI文档http://localhost:8000/docsSwagger UI搜索接口POST http://localhost:8000/api/crawler/search
