如何将asyncio GET请求实现为自包含类

改造思路 #

将原示例中的会话管理、单请求逻辑、批量任务调度全部封装到单个类中，实现完全自包含。为了避开__init__无法使用await的语法限制，选择用异步上下文管理器管理ClientSession的生命周期，同时对外暴露极简的调用入口。

改造后完整代码 #

import aiohttp
import asyncio

class PokemonFetcher:
    def __init__(self, start_id: int = 1, end_id: int = 151):
        # 可配置参数，外部初始化时可自定义爬取范围
        self.base_url = "https://pokeapi.co/api/v2/pokemon"
        self.start_id = start_id
        self.end_id = end_id
        self.session = None

    async def __aenter__(self):
        # 进入上下文时自动创建会话
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 退出上下文时自动关闭会话，避免资源泄露
        if self.session:
            await self.session.close()

    async def _get_single_pokemon(self, pokemon_id: int) -> str:
        # 内部私有方法：单条GET请求逻辑
        url = f"{self.base_url}/{pokemon_id}"
        async with self.session.get(url) as resp:
            pokemon_data = await resp.json()
            return pokemon_data["name"]

    async def fetch_all(self) -> list[str]:
        # 对外暴露的批量查询方法
        tasks = [
            asyncio.create_task(self._get_single_pokemon(pid))
            for pid in range(self.start_id, self.end_id + 1)
        ]
        return await asyncio.gather(*tasks)

async def main():
    # 外部调用逻辑非常简洁，不需要关心会话管理、任务组装等内部细节
    async with PokemonFetcher(1, 151) as fetcher:
        pokemon_names = await fetcher.fetch_all()
        for name in pokemon_names:
            print(name)

if __name__ == "__main__":
    asyncio.run(main())

关键设计说明 #

• 完全自包含：所有核心逻辑都封装在类内部，外部调用无需传递session等中间变量，仅需关注输入参数和返回结果
• 资源安全：通过异步上下文管理器自动管理会话的创建和销毁，不会出现资源泄露问题，也避开了构造函数无法使用await的语法限制
• 可扩展性强：如果需要新增请求头配置、超时设置、失败重试、结果字段调整等逻辑，仅需要修改类内部对应代码即可，不会影响外部调用逻辑

内容的提问来源于stack exchange，提问作者John Stud