AIOHTTP 流式分块下载

示例说明

resp.json() 或 resp.read() 会把整个响应体读进内存，下载大文件必然崩。iter_chunked 按块迭代响应流，每块到达即处理——这里顺带维护 sha256 摘要与进度统计，下载完成时校验值一并就绪。sock_read 超时只约束单次读取等待，整体不设限，长传输不会被误杀。

技术要点

• iter_chunked 按块消费响应流
• 哈希随块增量更新无需二次读取
• sock_read 只限制单次读取停滞

代码解读

函数在会话与响应两层上下文管理器里工作：先 raise_for_status 拒绝错误状态，再循环迭代块做三件事——更新摘要、累计字节、按块数节流打印进度。返回字典带字节数、块数与十六进制摘要，下载与校验一次完成。

使用建议

落盘场景把块直接写文件而不是累积。进度打印频率按块大小调节。哈希算法按对端约定选择，大文件可用分块校验协议。

代码示例

# stream chunks incrementally to disk instead of buffering
import asyncio
import hashlib

import aiohttp


async def download_and_hash(url: str, block_hint: int = 1024 * 64) -> dict:
    digest = hashlib.sha256()
    received = 0
    chunks = 0

    timeout = aiohttp.ClientTimeout(total=None, sock_read=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(url) as resp:
            resp.raise_for_status()
            total = int(resp.headers.get('Content-Length', 0))

            async for chunk in resp.content.iter_chunked(block_hint):
                digest.update(chunk)
                received += len(chunk)
                chunks += 1
                if chunks % 200 == 0:
                    print(f'progress {received / total:.1%}')

    return {
        'bytes': received,
        'chunks': chunks,
        'sha256': digest.hexdigest(),
    }


result = asyncio.run(
    download_and_hash('https://cdn.example.com/datasets/prices.parquet')
)
print('verified sha256:', result['sha256'][:16], 'bytes:', result['bytes'])