A Python-program that makes successive API-calls

Question

I wrote some code to make calls to a public API. The code is correct as required as usual. My goal is to make the code as self-explanatory, maintainable and robust as possible. I also tried to make the code extendable, such that it would be easy to add different API calls in the future. Execution speed or memory-efficiency are not really a big issue. I would appreciate any comments on my overall approach or implementation details. With my goals in mind: what could be improved?

The following code generates info about releases published on discogs.com. Every instance of ReleaseGenerator takes wantlist_url as a parameter, which is just the URL to a user generated list of releases. Release info can be retrieved like this:

url = 'https://api.discogs.com/users/damonrthomas/wants?page=1&per_page=500'
releases = ReleaseGenerator(url)
for release in releases:
    print(release)

Running the program outputs release info for all releases contained on the specified wantlist, e.g.:

Release(id=3010992, title='DJ-Kicks', artists=['Motor City Drum Ensemble'], genres=['Electronic', 'Jazz', 'Funk / Soul', 'Stage & Screen'], styles=['House', 'Fusion', 'Disco', 'Neo Soul', 'Deep House', 'Soundtrack', 'Minimal', 'Free Jazz'], labels=['!K7 Records'], year=2011, mediums=[{'medium': 'Vinyl', 'description': ['LP', 'Compilation'], 'qty': '2'}])
Release(id=8108236, title='DJ-Kicks', artists=['Moodymann'], genres=['Electronic', 'Hip Hop', 'Funk / Soul'], styles=['Soul', 'Downtempo', 'Deep House', 'House'], labels=['!K7 Records', '!K7 Records'], year=2016, mediums=[{'medium': 'Vinyl', 'description': ['LP', 'Compilation'], 'qty': '3'}])
Release(id=14917657, title='Sounds & Reasons', artists=['The Mighty Zaf', 'Phil Asher'], genres=['Funk / Soul'], styles=['Disco', 'Boogie'], labels=["'80s"], year=2020, mediums=[{'medium': 'Vinyl', 'description': ['12"', '33 ⅓ RPM', 'Test Pressing'], 'qty': '1'}])
Release(id=11771040, title='Genie', artists=['The Mighty Zaf', 'Phil Asher'], genres=['Funk / Soul'], styles=['Disco'], labels=["'80s"], year=2018, mediums=[{'medium': 'Vinyl', 'description': ['12"', '33 ⅓ RPM', 'Test Pressing'], 'qty': '1'}])
Release(id=12009859, title='Genie', artists=['The Mighty Zaf', 'Phil Asher'], genres=['Funk / Soul'], styles=['Disco'], labels=["'80s"], year=2018, mediums=[{'medium': 'Vinyl', 'description': ['12"', '33 ⅓ RPM'], 'qty': '1'}])
Release(id=16157292, title='Sounds & Reasons', artists=['The Mighty Zaf', 'Phil Asher'], genres=['Funk / Soul'], styles=['Disco', 'Boogie'], labels=["'80s"], year=2020, mediums=[{'medium': 'Vinyl', 'description': ['12"', '33 ⅓ RPM'], 'qty': '1'}])
...

The class ReleaseGenerator makes use of the public API of discogs.com and implements the retrieval of data:

import requests
from src.utils import verify
import time
from collections import namedtuple


class ReleaseGenerator:
    """Class for generating information about releases on the specified wantlist url"""
    def __init__(self, wantlist_url):
        self.wantlist_url = wantlist_url
        self.release_url = 'https://api.discogs.com/releases'
        self.want_key = 'wants'
        self.page_key = 'pagination'
        self.url_key = 'urls'
        self.id_key = 'id'
        self.next_key = 'next'

    @verify
    def _get_response(self, url):
        """Makes a request to the specified url."""
        request = requests.get(url)
        json = request.json()
        return json

    def _generate_pages(self):
        """Generates all pages of the paginated API response. Throttles requests automatically if needed."""
        url = self.wantlist_url
        while True:
            res = self._get_response(url)
            try:
                url = res[self.page_key][self.url_key][self.next_key]
            except KeyError:
                if 'message' in res.keys() and res['message'] == "You are making requests too quickly.":
                    time.sleep(60)
                else:
                    print('Warning: Unknown KeyError in parsing API response')
            else:
                if res[self.page_key]['page'] != res[self.page_key]['pages']:
                    yield res
                else:
                    break

    def _generate_releases_on_page(self, page):
        """Extracts needed information from API response page and returns it as a namedtuple."""
        Release = namedtuple('Release', 'id title artists genres styles labels year mediums')
        items = page[self.want_key]
        for item in items:
            release = Release(item[self.id_key],
                              item['basic_information']['title'],
                              [artist['name'] for artist in item['basic_information']['artists']],
                              [genre for genre in item['basic_information']['genres']],
                              [style for style in item['basic_information']['styles']],
                              [label['name'] for label in item['basic_information']['labels']],
                              item['basic_information']['year'],
                              [{'medium': medium['name'],
                                'description': medium['descriptions'],
                                'qty': medium['qty']} for medium in item['basic_information']['formats']])
            yield release

    def __iter__(self):
        """Generates release info for the specified wantlist url."""
        for page in self._generate_pages():
            for rid in self._generate_releases_on_page(page):
                yield rid

Finally, this is how the @verify decorator is implemented:

import requests
import sys
import time


def verify(fn):
    """
    Attempts to execute the function at most 10 times.
    If execution is successful, the function returns.
    If execution throws a Connection Error, try again after 10 seconds.
    If execution is not successful after 10 times, the program terminates.
    """
    def wrapper(*args, **kwargs):
        for attempt in range(10):
            try:
                json = fn(*args, **kwargs)
                break
            except requests.exceptions.ConnectionError:
                time.sleep(10)
        else:
            sys.exit()
        return json
    return wrapper

Answer 1

Sessions

When you are making repeated requests against a host, you should use requests.Session(). Sessions make it easy to persist cookies, headers etc and improve performance by reusing TCP connections. They should probably be the preferred way of using requests.

API

The API doc provides some useful information. For instance:

Your application must provide a User-Agent string that identifies itself – preferably something that follows RFC 1945...

But you are not doing that. So your user agent is a default value like: python-requests/2.26.0. Strike one.

They also write in the FAQ:

1. Why am I getting an empty response from the server?

This generally happens when you forget to add a User-Agent header to your requests.

Regarding rate limiting: they provide some headers in their responses, like this one:

X-Discogs-Ratelimit-Remaining: The number of remaining requests you are able to make in the existing rate limit window.

So you should be using those headers to track your own consumption and adjust your rate accordingly. Strike two.

Status code: you are not checking the status code for your requests. If you get anything else than 200 (or at least in the 200 range), than quite likely an error has occurred and you should log the response and investigate. The point is to make your bot aware, and not shoot blindly in the dark, spamming a server with requests that may not even succeed. Strike three.

If you get a non-200 response, then your code will crash anyway because you are attempting to fetch a JSON response that isn't there. It's easy to anticipate errors if you systematically inspect the status code.

You have a number of JSON keys like self.want_key = 'wants' or self.page_key = 'pagination' but I am not sure they provide much benefit because they are used in one function. They make the code slightly less transparent. instead of writing:

url = res[self.page_key][self.url_key][self.next_key]

I think you could simply write:

url = res['pagination']['urls']['next']

Basically, just like you did in function _generate_releases_on_page.

The API specifications may change at some point, but in either case you would have to perform search and replace across your code. I can't see how those variables improve the situation.

But you can protect yourself against unexpected changes in the format of the responses:

Currently, our API only supports one version: v2. However, you can specify a version in your requests to future-proof your application. By adding an Accept header with the version and media type, you can guarantee your requests will receive data from the correct version you develop your app on.

A standard Accept header may look like this: application/vnd.discogs.v2.html+json

Simplification

This function can be reduced a bit:

def _get_response(self, url):
    """Makes a request to the specified url."""
    request = requests.get(url)
    json = request.json()
    return json

Why not simply do:

def _get_response(self, url):
    """Makes a request to the specified url."""
    request = requests.get(url)
    return request.json()

You're not doing anything with the JSON, so the variable assignment is unneeded. This function is extremely short and it's very obvious what it does, so the comment does not teach me anything. But _get_response is very generic, I would perhaps rename the function to _get_json_response to be more explicit.

Note that the user agent and some other useful requests headers can be set once in your session, and then you no longer have to worry about them. They will be added automatically to your requests. Moreover, you can still pass additional headers on a per-request basis if needed.

while True is a code smell here:

def _generate_pages(self):
    """Generates all pages of the paginated API response. Throttles requests automatically if needed."""
    url = self.wantlist_url
    while True:
        res = self._get_response(url)
        try:
            url = res[self.page_key][self.url_key][self.next_key]
        except KeyError:
            if 'message' in res.keys() and res['message'] == "You are making requests too quickly.":
                time.sleep(60)
            else:
                print('Warning: Unknown KeyError in parsing API response')
        else:
            if res[self.page_key]['page'] != res[self.page_key]['pages']:
                yield res
            else:
                break

Since you expect a response like this:

{
  "pagination": {
    "page": 1,
    "pages": 17,
    "per_page": 500,
    "items": 8022,
    "urls": {
      "last": "https://api.discogs.com/users/damonrthomas/wants?page=17&per_page=500",
      "next": "https://api.discogs.com/users/damonrthomas/wants?page=2&per_page=500"
    }
  },
  "wants": [
...

you know how many pages to fetch. So you either do a regular for ... in range statement, or while current_page < pages.

The other problem is the timing, you are pounding the server with a lot of requests without pause - until you get blocked - and then you wait 60s to reset your window. It is a crude way of doing it.

I would either get rid of the verify decorator, or rewrite it to take full advantage of the response headers at your disposal. Then the function can be used to perform sensible throttling on your end, not just wait 60 seconds but a more flexible rate that is in line with the permitted usage.

You already have an indication here:

Requests are throttled by the server by source IP to 60 per minute for authenticated requests, and 25 per minute for unauthenticated requests, with some exceptions.

Get a token, then add a systematic delay of about 1s after every request + some margin and you should be fine. You may not experience any blocking if you play nice, and your code will perform better and more reliably. But do check the response headers to verify that you are always within limits. And do check the status code, always.