I have the following code that includes a simple class AioRequest
. I would like to be able to call an instance of this class directly with a single URL or a list of URLs and return the results.
request = AioRequest()
single_response = request('get', 'http://httpbin.org/get')
or
url_list = ['http://httpbin.org/get', 'http://google/some_json']
list_of_responses = request('get', url_list)
I'm wondering what the best practice is for two strategies I am using:
Determining if the argument is a list of URLs or a string with a single URL. Should it be two separate arguments like
__call__
or should I type check like__call__2
?I would like to use the
__call__
method as the main API for theAioRequest
instance, but is returning two different types based on the arguments given a good idea? Is is safe to rely on the user to determine the return type (example: give a single string get a single response, give a list of strings, get a list of responses)?
import asyncio
from aiohttp import ClientSession
from collections import namedtuple
Response = namedtuple(
"Response",
["url", "status_code", "json"]
)
class AioRequest:
def __init__(self, session=None):
self._loop = asyncio.get_event_loop()
self._session = session or ClientSession(loop=self._loop)
def __call__(self, method, url=None, url_list=None, **kwargs):
urls = url_list or []
urls.append(url)
requests = [self._request(method, url, **kwargs)
for url in urls
if url is not None]
return self._run(*requests)
def __call__2(self, method, url_or_list_of, **kwargs):
if type(url_or_list_of) is str:
return self._run(method, url_or_list_of, **kwargs)
else:
requests = [self._request(method, url, **kwargs)
for url in url_or_list_of]
return self._run(*requests)
async def _request(self, method, url, **kwargs):
response = await self._session.request(method, url, **kwargs)
return Response(url, response.status, await response.json())
def _run(self, *tasks):
task_group = asyncio.gather(*tasks)
results = self._loop.run_until_complete(task_group) # results = list of responses
if len(results) == 1:
return results[0]
return results
def __del__(self):
self._session.close()