一、背景与需求分析

企业工商信息查询是金融风控、供应链管理、市场调研等领域的核心需求。传统方式依赖人工查询或单一数据源，存在效率低、覆盖不全、更新滞后等问题。随着大数据技术的发展，集成多数据源（如天眼查、企查查、爱企查、国家企业信用信息公示系统）的API服务成为趋势，但面临技术架构选型、数据整合、并发处理等挑战。

FastAPI作为基于Python的现代Web框架，以高性能、异步支持、自动生成API文档等特性，成为构建高并发API服务的理想选择。Uvicorn作为ASGI服务器，与FastAPI无缝协作，支持异步请求处理，显著提升吞吐量。本文将详细介绍如何基于FastAPI与Uvicorn实现企业工商信息大数据查询系统，并集成多数据源。

二、技术选型与架构设计

1. 技术栈选择

• 后端框架：FastAPI（支持异步请求、自动生成OpenAPI文档）
• ASGI服务器：Uvicorn（高性能、支持异步）
• 数据源：天眼查API、企查查API、爱企查API、国家企业信用信息公示系统（爬虫或官方API）
• 数据库：MongoDB（存储查询记录、缓存结果）
• 缓存：Redis（加速重复查询）
• 异步任务队列：Celery（处理耗时操作，如爬虫）

2. 系统架构

系统分为四层：

1. API层：FastAPI接收查询请求，验证参数，调用服务层。
2. 服务层：整合多数据源查询逻辑，处理缓存、异步任务。
3. 数据层：调用第三方API或爬虫获取数据，存储至MongoDB。
4. 缓存层：Redis缓存高频查询结果，减少重复请求。

三、核心功能实现

1. FastAPI项目初始化

pip install fastapi uvicorn pymongo redis celery requests

创建main.py：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# 允许跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.get("/")
def read_root():
    return {"message": "企业工商信息查询API"}

2. 数据源集成

2.1 第三方API调用

以天眼查为例，封装请求逻辑：

import requests
from pydantic import BaseModel
class QueryParams(BaseModel):
    company_name: str
    credit_code: str = None
def query_tianyancha(params: QueryParams):
    url = "https://api.tianyancha.com/company/search"
    headers = {"Authorization": "Bearer YOUR_API_KEY"}
    payload = {"keyword": params.company_name}
    if params.credit_code:
        payload["credit_code"] = params.credit_code
    response = requests.get(url, headers=headers, params=payload)
    return response.json()

2.2 国家企业公示系统爬虫

使用requests和BeautifulSoup解析网页（需遵守robots协议）：

from bs4 import BeautifulSoup
def scrape_gsxt(company_name):
    url = f"http://www.gsxt.gov.cn/search?keyword={company_name}"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")
    # 解析企业列表，返回关键信息
    companies = []
    for item in soup.select(".company-item"):
        name = item.select_one(".name").text
        credit_code = item.select_one(".credit-code").text
        companies.append({"name": name, "credit_code": credit_code})
    return companies

3. 缓存与异步处理

3.1 Redis缓存

import redis
from fastapi import Depends
redis_client = redis.Redis(host="localhost", port=6379, db=0)
def get_cached_result(key: str):
    cached = redis_client.get(key)
    if cached:
        return cached.decode("utf-8")
    return None
def set_cached_result(key: str, value: str, expire: int = 3600):
    redis_client.setex(key, expire, value)

3.2 Celery异步任务

启动Celery worker：

celery -A tasks worker --loglevel=info

创建tasks.py：

from celery import shared_task
from .data_sources import query_tianyancha, scrape_gsxt
@shared_task
def async_query(company_name, credit_code=None):
    # 并行调用多数据源
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor() as executor:
        tianyancha_future = executor.submit(query_tianyancha, company_name, credit_code)
        gsxt_future = executor.submit(scrape_gsxt, company_name)
        tianyancha_result = tianyancha_future.result()
        gsxt_result = gsxt_future.result()
    return {"tianyancha": tianyancha_result, "gsxt": gsxt_result}

4. API路由设计

扩展main.py，添加查询路由：

from fastapi import HTTPException
from .tasks import async_query
from .cache import get_cached_result, set_cached_result
@app.post("/query")
async def query_company(params: QueryParams):
    cache_key = f"company:{params.company_name}:{params.credit_code or ''}"
    cached = get_cached_result(cache_key)
    if cached:
        return {"data": cached, "source": "cache"}
    # 调用异步任务
    task = async_query.delay(params.company_name, params.credit_code)
    result = task.get(timeout=10)  # 等待任务完成
    # 缓存结果
    set_cached_result(cache_key, str(result))
    return {"data": result, "source": "api"}

四、部署与优化

1. 使用Uvicorn部署

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

• --workers：根据CPU核心数设置，提升并发能力。
• 负载均衡：前端使用Nginx反向代理，分发请求至多个Uvicorn实例。

2. 性能优化

• 异步IO：FastAPI原生支持异步，避免阻塞操作。
• 连接池：使用motor（异步MongoDB驱动）和aioredis（异步Redis驱动）。
• 限流：通过slowapi或Nginx限制API调用频率，防止滥用。

3. 错误处理与日志

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )
@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response

五、总结与展望

本文介绍了基于FastAPI与Uvicorn的企业工商信息大数据查询系统实现方案，通过集成天眼查、企查查、爱企查及国家企业公示系统，提供了高效、可靠的API服务。关键点包括：

1. 异步架构：利用FastAPI的异步特性，提升并发处理能力。
2. 多数据源整合：封装第三方API和爬虫逻辑，实现数据聚合。
3. 缓存与异步任务：通过Redis和Celery优化性能，减少响应时间。
4. 可扩展性：模块化设计，便于新增数据源或功能。

未来可扩展方向：

• 增加更多数据源（如启信宝、企查猫）。
• 引入机器学习模型，分析企业风险。
• 支持GraphQL查询，灵活获取数据。

此方案适用于金融、法律、供应链等领域，为开发者提供了一套可复用的企业工商信息查询框架。