Python Async API Wrapper,如何构造它

问题描述 投票:0回答:1

目前我有一个 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 经常使用上下文管理器。

您可以向我指出任何采用类似方法的项目吗?

python-3.x asynchronous aiohttp
1个回答
0
投票

我会让你的类成为一个包装客户端会话的上下文管理器。例如,参见: https://github.com/mlowijs/tesla_api

那个只是在退出方法中调用

.close()
,但您甚至可以直接调用客户端上下文方法:

    async def __aenter__(self):
        await self._client.__aenter__()

    async def __aexit__(self, ...):
        await self._client.__aexit__(...)
© www.soinside.com 2019 - 2024. All rights reserved.