76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
# coding: utf-8
|
|
#
|
|
# Copyright (C) 2020, Team Kodi
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
"""Functions to interact with various web site APIs."""
|
|
|
|
from __future__ import absolute_import, unicode_literals
|
|
|
|
import json
|
|
# from pprint import pformat
|
|
try: #PY2 / PY3
|
|
from urllib2 import Request, urlopen
|
|
from urllib2 import URLError
|
|
from urllib import urlencode
|
|
except ImportError:
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import URLError
|
|
from urllib.parse import urlencode
|
|
try:
|
|
from typing import Text, Optional, Union, List, Dict, Any # pylint: disable=unused-import
|
|
InfoType = Dict[Text, Any] # pylint: disable=invalid-name
|
|
except ImportError:
|
|
pass
|
|
|
|
HEADERS = {}
|
|
|
|
|
|
def set_headers(headers):
|
|
HEADERS.update(headers)
|
|
|
|
|
|
def load_info(url, params=None, default=None, resp_type = 'json'):
|
|
# type: (Text, Optional[Dict[Text, Union[Text, List[Text]]]]) -> Union[dict, list]
|
|
"""
|
|
Load info from external api
|
|
|
|
:param url: API endpoint URL
|
|
:param params: URL query params
|
|
:default: object to return if there is an error
|
|
:resp_type: what to return to the calling function
|
|
:return: API response or default on error
|
|
"""
|
|
theerror = ''
|
|
if params:
|
|
url = url + '?' + urlencode(params)
|
|
req = Request(url, headers=HEADERS)
|
|
try:
|
|
response = urlopen(req)
|
|
except URLError as e:
|
|
if hasattr(e, 'reason'):
|
|
theerror = {'error': 'failed to reach the remote site\nReason: {}'.format(e.reason)}
|
|
elif hasattr(e, 'code'):
|
|
theerror = {'error': 'remote site unable to fulfill the request\nError code: {}'.format(e.code)}
|
|
if default is not None:
|
|
return default
|
|
else:
|
|
return theerror
|
|
if resp_type.lower() == 'json':
|
|
resp = json.loads(response.read().decode('utf-8'))
|
|
else:
|
|
resp = response.read().decode('utf-8')
|
|
return resp
|