I built this api wrapper based on https://github.com/mlowijs/tesla_api as suggested by a member of this community as an example/boiler plate on Python Async API Wrapper, how to structure it.
import asyncio
import aiohttp
class AsyncWrapper:
def __init__(self, api_key, useragent="API-Wrapper/0.2"):
self.url = f"https://example.api/v1"
self._api_key = api_key
self._useragent = useragent
self._headers = {"X-API-Key": self._api_key, "accept": "application/json", "User-Agent": self._useragent}
self._session = aiohttp.ClientSession()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()
async def close(self):
await self._session.close()
async def get_endpoint1(self, arg1, arg2):
endpoint = "/pair"
async with self._session.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2}, headers=self._headers) as response:
return await response.json()
async def get_endpoint2(self, arg1, arg2, optionalarg1=None, optionalarg2=None):
endpoint = "/token"
async with self._session.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2, "optionalarg1": optionalarg1, "optionalarg2": optionalarg2}, headers=self._headers) as response:
return await response.json()
The class is done using asynchronous context managers (__aenter__ __aexit__) to create and kill the session as shown here. However, a close() method is left there so it can be called to close the session in case the class wasn't instantiated using a context manager:
class AsyncWrapper:
def __init__(self, api_key, useragent="API-Wrapper/0.2"):
self.url = f"https://example.api/v1"
self._api_key = api_key
self._useragent = useragent
self._headers = {"X-API-Key": self._api_key, "accept": "application/json", "User-Agent": self._useragent}
self._session = aiohttp.ClientSession()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()
async def close(self):
await self._session.close()
When running this code initiating the object, and then closing the session explictly by using the close() method that I left in the class:
import asyncio
import AsyncWrapper
api_key = "my_key"
async def main():
client = AsyncWrapper(api_key)
response1 = await client.get_endpoint1("argument1", "argument2")
print(response1)
await client.close()
asyncio.run(main())
I get this output claiming unclosed client session. However, the session was explictly closed at the end. After this warning, the output proceeds as normal.
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x0000028D1F0E0390>
However, if I run the same code using a context manager. There's no "unclosed client session" warning:
import asyncio
import AsyncWrapper
api_key = "my_key"
async def main():
async with AsyncWrapper(api_key) as client:
response1 = await client.get_endpoint1("argument1", "argument2")
print(response1)
asyncio.run(main())