Skip to main content
Code Review

Return to Revisions

2 of 2
deleted 3 characters in body
Reinderien
  • 70.9k
  • 5
  • 76
  • 256

A Python-program that makes successive API-calls

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 1⁄3 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 1⁄3 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 1⁄3 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 1⁄3 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
lang-py

AltStyle によって変換されたページ (->オリジナル) /