2
\$\begingroup\$

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:

  1. 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?

  2. I would like to use the __call__ method as the main API for the AioRequest 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()
\$\endgroup\$
1
\$\begingroup\$

__call__ really confuses me. Type checking is the way to go for readability. On top of that, if you need to accept new types later, all you need to do is add an if statement.

In my opinion, if you absolutely want the brevity of __call__, accepting multiple types of arguments is fine, and so is returning multiple types (as is the case with many builtin and standard library functions). It is important that you mention this in the documentation, though.


Some other questionable things:

  • Instead of comparing type directly, you can use isinstance:

    if isinstance(url_or_list_of, str):
        ...
    
  • url_or_list_of is a somewhat unfriendly name for a public API. How about url_or_list_of_urls?

  • It's good that you provide a 'clean exit' method. __del__ isn't the right method to override, however, since the implementation is CPython-specific, and it may never actually be called. Instead, try defining __enter__ and __exit__:

    class AioRequest:
        ...
        def __enter__(self):
            return self
            # This should be all in most cases,
            # unless you want to do something with `self` before returning
        ...
        def __exit__(self, exc_type, exc_value, exc_traceback):
            self._session.close()
    

    You can then use with:

      with AioRequest() as request:
          # __enter__ called
          response = request( ... )
      # __exit__ called
    
| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.