HTTPX 持久缓存离线回放

示例说明

外部数据源不可用时业务不应完全停摆。缓存传输层把缓存判定做进请求链路：Cache-Control 头决定新鲜期，过期内直接回放缓存不再发网络请求；文件缓存让进程重启后热身依旧有效——先批量预热常用资源，断网时关键数据仍可读取，只影响更新鲜度。

技术要点

• 缓存判定遵循 Cache-Control 语义
• 文件缓存跨进程重启持久
• 预热任务批量建立缓存基线

代码解读

客户端用 CacheControlTransport 包装文件缓存，可缓存方法限定 GET，状态码白名单里 404 也缓存——负缓存防止对不存在资源的反复穿透。warm_cache 并发预热十个产品；主流程再次取用演示缓存命中，断网时同样能回放。

使用建议

敏感响应不要进磁盘缓存或对缓存目录加密。负缓存的 TTL 要短于正缓存。后台刷新配合 stale-while-revalidate 语义可以做到用户无感更新。

代码示例

import asyncio

import httpx
from httpx_cache import CacheControlTransport

client = httpx.AsyncClient(
    transport=CacheControlTransport(
        cache=httpx_cache.FileCache(cache_dir='.http_cache'),
        cacheable_methods=('GET',),
        cacheable_status_codes=(200, 301, 404),
    )
)


async def fetch_product(product_id: int) -> dict | None:
    try:
        resp = await client.get(f'https://api.example.com/products/{product_id}')
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPStatusError:
        return None


async def warm_cache(ids: list[int]) -> int:
    results = await asyncio.gather(
        *(fetch_product(pid) for pid in ids),
        return_exceptions=True,
    )
    return sum(1 for r in results if not isinstance(r, Exception))


async def main() -> None:
    await warm_cache(list(range(10)))

    # 断网场景：文件缓存仍可回放此前成功的响应
    product = await fetch_product(3)
    print('served from cache offline:', product is not None)
    await client.aclose()


asyncio.run(main())