Skip to content

Requests

RequestController

Handles API requests and manages cached responses for the LastFM API.

Source code in pylastfmapi/request.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class RequestController:
    """Handles API requests and manages cached responses for the LastFM API."""

    def __init__(
        self, user_agent: str, api_key: str, reset_cache: bool = False
    ) -> None:
        """Initializes the RequestController with user-agent and API key.

        Args:
            user_agent (str): The user-agent string to be sent with
                each request.
            api_key (str): The API key for authentication with the LastFM API.
            reset_cache (bool, optional): If True, clears the existing cache.
                Defaults to False.
        """
        self.headers = {'user-agent': user_agent}
        self.payload = {'api_key': api_key, 'format': 'json'}
        requests_cache.install_cache()
        if reset_cache:
            self.clear_cache()

    def request(self, payload: dict) -> T_Response:
        """Sends a request to the LastFM API and returns the response.

        Args:
            payload (dict): The query parameters for the request.

        Returns:
            T_Response: The HTTP response object, which may be cached.

        Raises:
            RequestErrorException: If the response status code is not
                200 (OK) or if the response contains an error.
        """
        self.payload.update(payload)
        response = requests.get(URL, headers=self.headers, params=self.payload)

        if response.status_code != HTTPStatus.OK:
            raise RequestErrorException(
                f'Something wrong, HTTP error {response.status_code}: '
                f'{response.text}'
            )

        content = response.json()
        if 'error' in content:
            raise RequestErrorException(
                f'Something wrong, error {content["error"]}: '
                f'{content["message"]}'
            )
        return response

    @staticmethod
    def clear_cache() -> None:
        """Clears the cache of stored API responses."""
        cache = requests_cache.get_cache()
        if cache:
            cache.clear()

    #########################################################################
    # PAGINATION
    #########################################################################

    def request_all_pages(
        self, payload: dict, parent_key: str, list_key: str, amount: int | None
    ) -> list[T_Response]:
        """Requests all pages of data from the API for a given query,
        handling pagination.

        Args:
            payload (dict): The query parameters for the request.
            parent_key (str): The parent key in the JSON response
                containing the desired data.
            list_key (str): The key within the parent key's value
                that contains the list of items.
            amount (int): The total number of items to request.

        Returns:
            list[T_Response]: A list of HTTP response objects,
                each representing a page of data.
        """
        responses = []
        page = 1
        num_pages = None

        payload['limit'] = LIMIT
        if amount:
            if amount < LIMIT:
                last_limit = amount
                num_pages = 1
            else:
                last_limit = amount % LIMIT
                num_pages = ceil(amount / LIMIT)

        while True:
            payload = {**payload, 'page': page}
            if num_pages == page:
                payload.update({'limit': last_limit})
            response = self.request(payload)
            content = response.json()

            if (
                len(content.get('taggings', content)[parent_key][list_key])
                == 0
            ):
                break
            if not getattr(response, 'from_cache', False):
                time.sleep(0.25)
            responses.append(response)

            if 'taggings' in content:
                if page == int(content['taggings']['@attr']['totalPages']):
                    break
            elif page == int(content[parent_key]['@attr']['totalPages']):
                break

            if num_pages and page == num_pages:
                break

            page += 1
        return responses

    def get_paginated_data(
        self, payload: dict, parent_key: str, list_key: str, amount: int | None
    ) -> list[dict]:
        """Fetches paginated data from the LastFM API based on the
        given parameters.

        This method handles the pagination of API requests to gather a
        specified amount
        of data, combining the results into a single list.

        Args:
            payload (dict): The parameters to send to the API.
            parent_key (str): The key in the API response that contains the
                primary data structure.
            list_key (str): The key within the `parent_key` that contains
                the list of items.
            amount (int): The total number of elements to retrieve from
                the API.

        Returns:
            list[dict]: A list of dictionaries containing the retrieved data.
        """
        responses = self.request_all_pages(
            payload, parent_key, list_key, amount
        )
        response_list = []
        for index, data in enumerate(responses, start=1):
            _data = data.json()
            if amount and index == len(responses):
                left_data = amount % LIMIT_SEARCH
                response_list.extend(
                    _data.get('taggings', _data)[parent_key][list_key][
                        :left_data
                    ]
                )
            else:
                response_list.extend(
                    _data.get('taggings', _data)[parent_key][list_key]
                )
        return response_list

    #########################################################################
    # SEARCHES
    #########################################################################

    def request_search_pages(
        self, payload: dict, parent_key: str, list_key: str, amount: int | None
    ) -> list[T_Response]:
        """Requests all pages of data from the API for a given query,
        handling pagination. Specific for LastFM search format.

        Args:
            payload (dict): The query parameters for the search request.
            parent_key (str): The parent key in the JSON response
                containing the search results.
            list_key (str): The key within the parent key's value
                that contains the list of items.
            amount (int): The total number of items to request.

        Returns:
            list[T_Response]: A list of HTTP response objects,
                each representing a page of search results.
        """
        responses = []
        page = 1
        num_pages = None

        payload['limit'] = LIMIT_SEARCH
        if amount:
            if amount < LIMIT_SEARCH:
                num_pages = 1
                payload['limit'] = amount
            else:
                num_pages = ceil(amount / LIMIT_SEARCH)

        while True:
            payload = {**payload, 'page': page}
            response = self.request(payload)
            content = response.json()

            if len(content['results'][parent_key][list_key]) == 0:
                break
            if not getattr(response, 'from_cache', False):
                time.sleep(0.25)
            responses.append(response)

            if page == ceil(
                int(content['results']['opensearch:totalResults'])
                / LIMIT_SEARCH
            ):
                break
            if num_pages and page == num_pages:
                break

            page += 1
        return responses

    def get_search_data(
        self, payload: dict, parent_key: str, list_key: str, amount: int | None
    ) -> list[dict]:
        """Fetches search result data from the LastFM API based on the
        given parameters.

        This method handles the pagination of search result requests,
        retrieving
        the specified number of results and combining them into a single list.

        Args:
            payload (dict): The parameters to send to the API for the
                search request.
            parent_key (str): The key in the API response that contains
                the primary data structure.
            list_key (str): The key within the `parent_key` that contains
                the list of items.
            amount (int): The total number of elements to retrieve from
                the API.

        Returns:
            list[dict]: A list of dictionaries containing the search results.
        """
        responses = self.request_search_pages(
            payload, parent_key, list_key, amount
        )
        response_list = []
        for index, data in enumerate(responses, start=1):
            _data = data.json()
            if amount and index == len(responses):
                left_data = amount % LIMIT_SEARCH
                response_list.extend(
                    _data['results'][parent_key][list_key][:left_data]
                )
            else:
                response_list.extend(_data['results'][parent_key][list_key])
        return response_list

__init__(user_agent, api_key, reset_cache=False)

Initializes the RequestController with user-agent and API key.

Parameters:

Name Type Description Default
user_agent str

The user-agent string to be sent with each request.

 required
api_key str

The API key for authentication with the LastFM API.

 required
reset_cache bool

If True, clears the existing cache. Defaults to False.

 False
Source code in pylastfmapi/request.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self, user_agent: str, api_key: str, reset_cache: bool = False
) -> None:
    """Initializes the RequestController with user-agent and API key.

    Args:
        user_agent (str): The user-agent string to be sent with
            each request.
        api_key (str): The API key for authentication with the LastFM API.
        reset_cache (bool, optional): If True, clears the existing cache.
            Defaults to False.
    """
    self.headers = {'user-agent': user_agent}
    self.payload = {'api_key': api_key, 'format': 'json'}
    requests_cache.install_cache()
    if reset_cache:
        self.clear_cache()

clear_cache() staticmethod

Clears the cache of stored API responses.

Source code in pylastfmapi/request.py 
74
75
76
77
78
79
@staticmethod
def clear_cache() -> None:
    """Clears the cache of stored API responses."""
    cache = requests_cache.get_cache()
    if cache:
        cache.clear()

get_paginated_data(payload, parent_key, list_key, amount)

Fetches paginated data from the LastFM API based on the given parameters.

This method handles the pagination of API requests to gather a specified amount of data, combining the results into a single list.

Parameters:

Name Type Description Default
payload dict

The parameters to send to the API.

 required
parent_key str

The key in the API response that contains the primary data structure.

 required
list_key str

The key within the parent_key that contains the list of items.

 required
amount int

The total number of elements to retrieve from the API.

 required

Returns:

Type Description
list[dict]

list[dict]: A list of dictionaries containing the retrieved data.
Source code in pylastfmapi/request.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def get_paginated_data(
    self, payload: dict, parent_key: str, list_key: str, amount: int | None
) -> list[dict]:
    """Fetches paginated data from the LastFM API based on the
    given parameters.

    This method handles the pagination of API requests to gather a
    specified amount
    of data, combining the results into a single list.

    Args:
        payload (dict): The parameters to send to the API.
        parent_key (str): The key in the API response that contains the
            primary data structure.
        list_key (str): The key within the `parent_key` that contains
            the list of items.
        amount (int): The total number of elements to retrieve from
            the API.

    Returns:
        list[dict]: A list of dictionaries containing the retrieved data.
    """
    responses = self.request_all_pages(
        payload, parent_key, list_key, amount
    )
    response_list = []
    for index, data in enumerate(responses, start=1):
        _data = data.json()
        if amount and index == len(responses):
            left_data = amount % LIMIT_SEARCH
            response_list.extend(
                _data.get('taggings', _data)[parent_key][list_key][
                    :left_data
                ]
            )
        else:
            response_list.extend(
                _data.get('taggings', _data)[parent_key][list_key]
            )
    return response_list

get_search_data(payload, parent_key, list_key, amount)

Fetches search result data from the LastFM API based on the given parameters.

This method handles the pagination of search result requests, retrieving the specified number of results and combining them into a single list.

Parameters:

Name Type Description Default
payload dict

The parameters to send to the API for the search request.

 required
parent_key str

The key in the API response that contains the primary data structure.

 required
list_key str

The key within the parent_key that contains the list of items.

 required
amount int

The total number of elements to retrieve from the API.

 required

Returns:

Type Description
list[dict]

list[dict]: A list of dictionaries containing the search results.
Source code in pylastfmapi/request.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def get_search_data(
    self, payload: dict, parent_key: str, list_key: str, amount: int | None
) -> list[dict]:
    """Fetches search result data from the LastFM API based on the
    given parameters.

    This method handles the pagination of search result requests,
    retrieving
    the specified number of results and combining them into a single list.

    Args:
        payload (dict): The parameters to send to the API for the
            search request.
        parent_key (str): The key in the API response that contains
            the primary data structure.
        list_key (str): The key within the `parent_key` that contains
            the list of items.
        amount (int): The total number of elements to retrieve from
            the API.

    Returns:
        list[dict]: A list of dictionaries containing the search results.
    """
    responses = self.request_search_pages(
        payload, parent_key, list_key, amount
    )
    response_list = []
    for index, data in enumerate(responses, start=1):
        _data = data.json()
        if amount and index == len(responses):
            left_data = amount % LIMIT_SEARCH
            response_list.extend(
                _data['results'][parent_key][list_key][:left_data]
            )
        else:
            response_list.extend(_data['results'][parent_key][list_key])
    return response_list

request(payload)

Sends a request to the LastFM API and returns the response.

Parameters:

Name Type Description Default
payload dict

The query parameters for the request.

 required

Returns:

Name Type Description
T_Response T_Response

The HTTP response object, which may be cached.

Raises:

Type Description
RequestErrorException

If the response status code is not 200 (OK) or if the response contains an error.
Source code in pylastfmapi/request.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def request(self, payload: dict) -> T_Response:
    """Sends a request to the LastFM API and returns the response.

    Args:
        payload (dict): The query parameters for the request.

    Returns:
        T_Response: The HTTP response object, which may be cached.

    Raises:
        RequestErrorException: If the response status code is not
            200 (OK) or if the response contains an error.
    """
    self.payload.update(payload)
    response = requests.get(URL, headers=self.headers, params=self.payload)

    if response.status_code != HTTPStatus.OK:
        raise RequestErrorException(
            f'Something wrong, HTTP error {response.status_code}: '
            f'{response.text}'
        )

    content = response.json()
    if 'error' in content:
        raise RequestErrorException(
            f'Something wrong, error {content["error"]}: '
            f'{content["message"]}'
        )
    return response

request_all_pages(payload, parent_key, list_key, amount)

Requests all pages of data from the API for a given query, handling pagination.

Parameters:

Name Type Description Default
payload dict

The query parameters for the request.

 required
parent_key str

The parent key in the JSON response containing the desired data.

 required
list_key str

The key within the parent key's value that contains the list of items.

 required
amount int

The total number of items to request.

 required

Returns:

Type Description
list[T_Response]

list[T_Response]: A list of HTTP response objects, each representing a page of data.
Source code in pylastfmapi/request.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def request_all_pages(
    self, payload: dict, parent_key: str, list_key: str, amount: int | None
) -> list[T_Response]:
    """Requests all pages of data from the API for a given query,
    handling pagination.

    Args:
        payload (dict): The query parameters for the request.
        parent_key (str): The parent key in the JSON response
            containing the desired data.
        list_key (str): The key within the parent key's value
            that contains the list of items.
        amount (int): The total number of items to request.

    Returns:
        list[T_Response]: A list of HTTP response objects,
            each representing a page of data.
    """
    responses = []
    page = 1
    num_pages = None

    payload['limit'] = LIMIT
    if amount:
        if amount < LIMIT:
            last_limit = amount
            num_pages = 1
        else:
            last_limit = amount % LIMIT
            num_pages = ceil(amount / LIMIT)

    while True:
        payload = {**payload, 'page': page}
        if num_pages == page:
            payload.update({'limit': last_limit})
        response = self.request(payload)
        content = response.json()

        if (
            len(content.get('taggings', content)[parent_key][list_key])
            == 0
        ):
            break
        if not getattr(response, 'from_cache', False):
            time.sleep(0.25)
        responses.append(response)

        if 'taggings' in content:
            if page == int(content['taggings']['@attr']['totalPages']):
                break
        elif page == int(content[parent_key]['@attr']['totalPages']):
            break

        if num_pages and page == num_pages:
            break

        page += 1
    return responses

request_search_pages(payload, parent_key, list_key, amount)

Requests all pages of data from the API for a given query, handling pagination. Specific for LastFM search format.

Parameters:

Name Type Description Default
payload dict

The query parameters for the search request.

 required
parent_key str

The parent key in the JSON response containing the search results.

 required
list_key str

The key within the parent key's value that contains the list of items.

 required
amount int

The total number of items to request.

 required

Returns:

Type Description
list[T_Response]

list[T_Response]: A list of HTTP response objects, each representing a page of search results.
Source code in pylastfmapi/request.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def request_search_pages(
    self, payload: dict, parent_key: str, list_key: str, amount: int | None
) -> list[T_Response]:
    """Requests all pages of data from the API for a given query,
    handling pagination. Specific for LastFM search format.

    Args:
        payload (dict): The query parameters for the search request.
        parent_key (str): The parent key in the JSON response
            containing the search results.
        list_key (str): The key within the parent key's value
            that contains the list of items.
        amount (int): The total number of items to request.

    Returns:
        list[T_Response]: A list of HTTP response objects,
            each representing a page of search results.
    """
    responses = []
    page = 1
    num_pages = None

    payload['limit'] = LIMIT_SEARCH
    if amount:
        if amount < LIMIT_SEARCH:
            num_pages = 1
            payload['limit'] = amount
        else:
            num_pages = ceil(amount / LIMIT_SEARCH)

    while True:
        payload = {**payload, 'page': page}
        response = self.request(payload)
        content = response.json()

        if len(content['results'][parent_key][list_key]) == 0:
            break
        if not getattr(response, 'from_cache', False):
            time.sleep(0.25)
        responses.append(response)

        if page == ceil(
            int(content['results']['opensearch:totalResults'])
            / LIMIT_SEARCH
        ):
            break
        if num_pages and page == num_pages:
            break

        page += 1
    return responses