discogs_client.fetchers module#

class discogs_client.fetchers.Fetcher#

Bases: object

Base class for Fetchers, which wrap and normalize the APIs of various HTTP libraries.

(It’s a slightly leaky abstraction designed to make testing easier.)

backoff_enabled = True#
connect_timeout: float | int | None = None#
read_timeout: float | int | None = None#
fetch(client, method, url, data=None, headers=None, json=True)#

Fetch the given request

Parameters:

  • client (object) – Instantiated discogs_client.client.Client object.

  • method (str) – HTTP method.

  • url (str) – API endpoint URL.

  • data (dict, optional) – data to be sent in the request’s body, by default None.

  • headers (dict, optional) – HTTP headers, by default None.

  • json_format (bool, optional) – If True, an object passed with the “data” arg will be converted into a JSON string, by default True.

Returns:

  • content (bytes) – as returned by Python “Requests”

  • status_code (int) – as returned by Python “Requests”

Raises:

NotImplementedError – Is raised if a child class doesn’t implement a fetch method.

request(method, url, data, headers, params=None)#
class discogs_client.fetchers.LoggingDelegator(fetcher)#

Bases: object

Wraps a fetcher and logs all requests.

property last_request#
fetch(client, method, url, data=None, headers=None, json=True)#

Appends passed “fetcher” to a requests list and returns result of fetcher.fetch method

class discogs_client.fetchers.RequestsFetcher#

Bases: Fetcher

Fetches via HTTP from the Discogs API (unauthenticated)

fetch(client, method, url, data=None, headers=None, json=True)#
Parameters:

  • client (object) – Unused in this subclass.

  • method (str) – HTTP method.

  • url (str) – API endpoint URL.

  • data (dict, optional) – data to be sent in the request’s body, by default None.

  • headers (dict, optional) – HTTP headers, by default None.

  • json_format (bool, optional) – Unused in this subclass, by default True.

Returns:

  • content (bytes) – as returned by Python “Requests”

  • status_code (int) – as returned by Python “Requests”

class discogs_client.fetchers.UserTokenRequestsFetcher(user_token)#

Bases: Fetcher

Fetches via HTTP from the Discogs API using User-token authentication

fetch(client, method, url, data=None, headers=None, json_format=True)#

Fetch the given request on the user’s behalf

Parameters:

  • client (object) – Unused in this subclass.

  • method (str) – HTTP method.

  • url (str) – API endpoint URL.

  • data (dict, optional) – data to be sent in the request’s body, by default None.

  • headers (dict, optional) – HTTP headers, by default None.

  • json_format (bool, optional) – If True, an object passed with the “data” arg will be converted into a JSON string, by default True.

Returns:

  • content (bytes) – as returned by Python “Requests”

  • status_code (int) – as returned by Python “Requests”

class discogs_client.fetchers.OAuth2Fetcher(consumer_key, consumer_secret, token=None, secret=None)#

Bases: Fetcher

Fetches via HTTP + OAuth 1.0a from the Discogs API.

store_token_from_qs(query_string)#
forget_token()#
store_token(token, secret)#
set_verifier(verifier)#
fetch(client, method, url, data=None, headers=None, json_format=True)#

Fetch the given request on the user’s behalf

Parameters:

  • client (object) – Unused in this subclass.

  • method (str) – HTTP method.

  • url (str) – API endpoint URL.

  • data (dict, optional) – Data to be sent in the request’s body, by default None.

  • headers (dict, optional) – HTTP headers, by default None.

  • json_format (bool, optional) – If True, an object passed with the “data” arg will be converted into a JSON string, by default True.

Returns:

  • content (bytes) – as returned by Python “Requests”

  • status_code (int) – as returned by Python “Requests”

class discogs_client.fetchers.FilesystemFetcher(base_path)#

Bases: Fetcher

Fetches from a directory of files.

default_response = ('{"message": "Resource not found."}', 404)#
path_with_params = re.compile('(?P<dir>(\\w+/)+)(?P<query>\\w+)\\?(?P<params>.*)')#
fetch(client, method, url, data=None, headers=None, json=True)#

Fetch the given request

Parameters:

  • client (object) – Instantiated discogs_client.client.Client object.

  • method (str) – Unused in this subclass (this fetcher supports GET only).

  • url (str) – API endpoint URL.

  • data (dict, optional) – Unused in this subclass (this fetcher supports GET only).

  • headers (dict, optional) – Unused in this subclass.

  • json_format (bool, optional)

Returns:

  • content (bytes)

  • status_code (int)

check_alternate_params(base_name, json)#

parse_qs() result is non-deterministic - a different file might be requested, making the tests fail randomly, depending on the order of parameters in the query. This fixes it by checking for matching file names with a different permutations of the parameters.

class discogs_client.fetchers.MemoryFetcher(responses)#

Bases: Fetcher

Fetches from a dict of URL -> (content, status_code).

default_response = ('{"message": "Resource not found."}', 404)#
fetch(client, method, url, data=None, headers=None, json=True)#

Fetch the given request

Parameters:

  • client (object) – Unused in this subclass.

  • method (str) – Unused in this subclass (this fetcher supports GET only).

  • url (str) – API endpoint URL.

  • data (dict, optional) – Unused in this subclass (this fetcher supports GET only).

  • headers (dict, optional) – Unused in this subclass.

  • json_format (bool, optional) – Unused in this subclass

Returns:

  • content (bytes)

  • status_code (int)