如何基于aiohttp构建Python异步API封装类？求参考项目

问题描述 #

我有一个API服务封装项目，当前用可实例化类通过类方法访问所有接口，返回Python字典。示例代码如下：

class APIProject:
    def __init__(self, api_key, useragent="MyUserAgent"):
        self.url = f"https://api.server.io/v1"
        self._api_key = api_key
        self._useragent = useragent
        self._headers = {"X-API-Key": self._api_key, "accept": "application/json", "User-Agent": self._useragent}

    def get_endpoint1(self, arg1, arg2):
        endpoint = "/endpoint1"
        response = requests.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2}, headers=self._headers)
        return response.json()

    def get_endpoint2(self, arg1, arg2, arg3=None, arg4=None):
        endpoint = "/endpoint2"
        response = requests.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2, "arg3": arg3, "arg4": arg4}, headers=self._headers)
        return response.json()

希望用aiohttp实现异步版本，如何沿用类似类结构？另外aiohttp常用上下文管理器，求推荐同类实现的参考项目。

异步版本实现方案 #

可以完全沿用类结构，将类方法改为异步方法，同时在类内部管理aiohttp的ClientSession——这是aiohttp推荐的复用方式，自带连接池，比每次请求新建会话更高效。

实现代码 #

import aiohttp

class AsyncAPIProject:
    def __init__(self, api_key, useragent="MyUserAgent"):
        self.base_url = "https://api.server.io/v1"
        self._headers = {
            "X-API-Key": api_key,
            "accept": "application/json",
            "User-Agent": useragent
        }
        self._session = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession(headers=self._headers)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._session.close()

    async def get_endpoint1(self, arg1, arg2):
        if not self._session:
            raise RuntimeError("请通过async with上下文管理器使用该类实例")
        url = f"{self.base_url}/endpoint1"
        async with self._session.get(url, params={"arg1": arg1, "arg2": arg2}) as response:
            response.raise_for_status()
            return await response.json()

    async def get_endpoint2(self, arg1, arg2, arg3=None, arg4=None):
        if not self._session:
            raise RuntimeError("请通过async with上下文管理器使用该类实例")
        url = f"{self.base_url}/endpoint2"
        params = {"arg1": arg1, "arg2": arg2, "arg3": arg3, "arg4": arg4}
        filtered_params = {k: v for k, v in params.items() if v is not None}
        async with self._session.get(url, params=filtered_params) as response:
            response.raise_for_status()
            return await response.json()

使用示例 #

import asyncio

async def main():
    async with AsyncAPIProject(api_key="your_api_key") as api_client:
        result1 = await api_client.get_endpoint1("value1", "value2")
        result2 = await api_client.get_endpoint2("value1", "value2", arg3="value3")
        print("Endpoint1 result:", result1)
        print("Endpoint2 result:", result2)

asyncio.run(main())

核心要点 #

• 类实现__aenter__和__aexit__方法，支持async with上下文管理器，确保会话正确关闭，避免资源泄漏。
• 所有接口方法定义为async def，调用时必须加await关键字。
• 加入response.raise_for_status()主动抛出HTTP错误，便于捕获和处理请求异常。
• 过滤参数中的None值，避免向API传递无效参数。

参考项目推荐 #

• aiohttp-requests：简化aiohttp调用逻辑，提供类requests的API风格，底层复用ClientSession，适合快速从requests迁移到异步场景。
• discord.py：Discord官方异步API封装库，完全基于类结构和aiohttp，大量使用上下文管理器管理会话，代码结构清晰，是学习异步API封装的典型案例。
• aiogram：Telegram Bot异步开发框架，基于aiohttp构建，类设计合理，涵盖多种异步API调用模式，能学习到异步资源管理的实用技巧。

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