用 FastAPI 封装一个高可用代理获取接口服务-ip资讯- kookeey

随着 Web 数据规模增长，越来越多 Python 开发者、爬虫工程师、后端服务开发者需要在业务中使用 HTTP 代理服务 来突破限制、模拟真实用户访问、实现高序列抓取等目标。本文将从实战角度出发，详细讲解如何用 FastAPI 封装一个清晰可维护的代理获取接口服务，让你的爬虫或微服务能够“透明使用代理”，而无需频繁改写调用逻辑。

文章适合人群：

熟悉 Python 开发、Flask/Django 其中一的后端工程师
需要构建 API 接口以对外提供代理服务的开发者
构建爬虫框架，希望对代理 IP 调用进行统一管理者

示例中我们将构建一个可扩展的服务，它具备：

标准 REST API 接口用于获取代理
基于中间件 / 依赖注入管理 request
将代理请求逻辑和业务层解耦，提高可维护性
兼容真实住宅 IP 代理方案（如 kookeey 静态住宅代理 & 动态住宅代理）

一、核心功能：FastAPI如何封装代理获取接口？

本节带你从零开始搭建一个可运行的代理接口服务，让任意业务只需调用一个 API，就能自动通过代理发起 HTTP 请求。

项目初始化

mkdir fastapi-proxy-service

cd fastapi-proxy-service

python3 -m venv venv

source venv/bin/activate

pip install fastapi uvicorn httpx pydantic

定义数据模型 schema.py

from pydantic import BaseModel

class ProxyRequest(BaseModel):

    url: str

    method: str = "GET"

    timeout: int = 10

class ProxyResponse(BaseModel):

    status_code: int

    content: str

    proxy_used: str

    error: str = ""

封装代理逻辑 proxy_manager.py

import httpx, random

PROXY_POOL = [

    "http://127.0.0.1:8001",

    "http://127.0.0.1:8002"

]

def choose_proxy():

    return random.choice(PROXY_POOL)

async def fetch_via_proxy(url, method="GET", timeout=10):

    proxy = choose_proxy()

    proxies = {"http://": proxy, "https://": proxy}

    async with httpx.AsyncClient(proxies=proxies, timeout=timeout) as client:

        resp = await client.request(method, url)

        return {

            "status_code": resp.status_code,

            "content": resp.text,

            "proxy_used": proxy

        }

业务层 service.py

from fastapi import HTTPException

from app.schema import ProxyRequest, ProxyResponse

from app.proxy_manager import fetch_via_proxy

async def handle_proxy_request(req: ProxyRequest) -> ProxyResponse:

    try:

        result = await fetch_via_proxy(req.url, req.method, req.timeout)

        return ProxyResponse(**result)

    except Exception as e:

        raise HTTPException(status_code=500, detail=str(e))

路由入口 main.py

from fastapi import FastAPI

from app.schema import ProxyRequest, ProxyResponse

from app.service import handle_proxy_request

app = FastAPI()

@app.post("/proxy", response_model=ProxyResponse)

async def proxy(req: ProxyRequest):

    return await handle_proxy_request(req)

至此，一个基础可运行的代理 API 服务已经完成。

二、代理能力升级：接入高质量住宅代理与真实业务场景

要让代理服务真正能在业务中落地，仅靠本地代理池是不够的。本节将从以下几个方面来介绍：为什么要引入专业代理、不同代理类型如何选择以及kookeey 如何与系统无缝集成。

为什么要引入专业代理？

普通代理常见问题：

IP 来源不稳定且易被封禁
匿名度低，无法模拟真实用户
高并发下极易崩溃

相比之下，住宅代理最接近真实用户流量，是多数企业采用的标准方案。kookeey 的动态住宅代理支持全球 41 个以上的国家和地区，可实现精准的 IP 控制，并能适应不同的业务需求。点击下方链接即可免费试用：

kookeey 全球代理IP点击按钮免费试用

免费试用立即注册

动态住宅代理：自动轮换接入示例

示例代码（不暴露真实 API）：

async def fetch_kookeey_ip():

    resp = await httpx.get(

        "https://api.kookeey.com/residential/dynamic/get-ip",

        params={"country": "US", "count": 1}

    )

    return resp.json()["ip_list"]

加入代理池：

async def refill_pool():

    new_ips = await fetch_kookeey_ip()

    PROXY_POOL.extend(f"http://{ip}" for ip in new_ips)

静态住宅代理：绑定式业务接入

STATIC_PROXY = "http://username:password@gateway.kookeey.com:12345"

async def fetch_static(url):

    async with httpx.AsyncClient(proxies=STATIC_PROXY) as client:

        return (await client.get(url)).text

最佳实践：依赖注入代理

async def get_proxy():

    return choose_proxy()

@app.post("/proxy")

async def proxy(req: ProxyRequest, proxy = Depends(get_proxy)):

    return await fetch_via_proxy(req.url, method=req.method, timeout=req.timeout)

三、工程化优化：让服务足以支撑生产环境

性能优化：连接池与生命周期

app = FastAPI()

@app.on_event("startup")

async def startup():

    app.state.client = httpx.AsyncClient(

        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)

    )

@app.on_event("shutdown")

async def shutdown():

    await app.state.client.aclose()

代理池高可用：健康检查与自愈

async def health_check():

    while True:

        for proxy in list(PROXY_POOL):

            try:

                await app.state.client.get("https://httpbin.org/ip", proxies={"http": proxy})

            except:

                PROXY_POOL.remove(proxy)

        await asyncio.sleep(30)

架构升级：从工具到代理网关

当你的 FastAPI 服务接入企业应用后，它会进化为“代理网关”，具备：

统一代理策略。
多业务复用能力。
可观测性与访问追踪。
可扩展的调度能力（不同业务走不同代理池）。

四、生产落地：日志、追踪 ID 与鉴权

要上线到正式环境，这些能力必不可少。

统一日志

logger.info({

  "event": "proxy_request",

  "url": req.url,

  "proxy": result.proxy_used,

  "status": result.status_code

})

注入请求 ID

@app.middleware("http")

async def add_request_id(request, call_next):

    request.state.request_id = str(uuid.uuid4())

    response = await call_next(request)

    response.headers["X-Request-ID"] = request.state.request_id

    return response

API 鉴权

def verify_token(token: str):

    if token != "YOUR_TOKEN":

        raise HTTPException(status_code=401, detail="Invalid token")

结语：从简单代理接口到企业级代理平台

通过本次重构，你的 FastAPI 代理服务已经从一个“可运行的小工具”，成长为一个具备：

高可用代理池
动态住宅代理自动轮换
企业级性能优化
生产级日志与鉴权体系

特别是在集成 kookeey 的静态住宅代理与动态住宅代理后，服务在稳定性、成功率、跨地域访问能力方面会得到质的提升，完全能够支撑企业级业务场景。

本文来自网络投稿，不代表kookeey立场，如有问题请联系我们

用 FastAPI 封装一个高可用代理获取接口服务