refactorapi-client): fix linting errors

This commit is contained in:
jo 2022-08-09 17:16:48 +02:00 committed by Kyle Robbertze
parent 313898137c
commit 5782e10250
3 changed files with 33 additions and 33 deletions

View File

@ -16,6 +16,7 @@ class UrlException(Exception):
class IncompleteUrl(UrlException):
def __init__(self, url):
super().__init__()
self.url = url
def __str__(self):
@ -24,6 +25,7 @@ class IncompleteUrl(UrlException):
class UrlBadParam(UrlException):
def __init__(self, url, param):
super().__init__()
self.url = url
self.param = param
@ -31,6 +33,7 @@ class UrlBadParam(UrlException):
return f"Bad param '{self.param}' passed into url: '{self.url}'"
# pylint: disable=too-few-public-methods
class KeyAuth(AuthBase):
def __init__(self, key):
self.key = key
@ -49,9 +52,9 @@ class ApcUrl:
def params(self, **params):
temp_url = self.base_url
for k, v in params.items():
for k in params:
wrapped_param = "{" + k + "}"
if not wrapped_param in temp_url:
if wrapped_param not in temp_url:
raise UrlBadParam(self.base_url, k)
temp_url = temp_url.format_map(UrlParamDict(**params))
return ApcUrl(temp_url)
@ -59,7 +62,6 @@ class ApcUrl:
def url(self):
if "{" in self.base_url:
raise IncompleteUrl(self.base_url)
else:
return self.base_url
@ -123,13 +125,13 @@ class ApiRequest:
self.__req = lambda: self(*args, **kwargs)
return self
def retry(self, n, delay=5):
def retry(self, count, delay=5):
"""Try to send request n times. If after n times it fails then
we finally raise exception"""
for i in range(0, n - 1):
for _ in range(0, count - 1):
try:
return self.__req()
except Exception:
except requests.exceptions.RequestException:
sleep(delay)
return self.__req()
@ -161,5 +163,5 @@ class RequestProvider:
def __getattr__(self, attr):
if attr in self:
return self.requests[attr]
else:
return super().__getattribute__(attr)

View File

@ -93,23 +93,23 @@ class ApiClient:
return -1
def is_server_compatible(self, verbose=True):
logger = self.logger
api_version = self.__get_api_version()
if api_version == -1:
if verbose:
logger.info("Unable to get Airtime API version number.\n")
self.logger.info("Unable to get Airtime API version number.\n")
return False
elif api_version[0:3] != AIRTIME_API_VERSION[0:3]:
if api_version[0:3] != AIRTIME_API_VERSION[0:3]:
if verbose:
logger.info("Airtime API version found: " + str(api_version))
logger.info(
self.logger.info("Airtime API version found: " + str(api_version))
self.logger.info(
"pypo is only compatible with API version: " + AIRTIME_API_VERSION
)
return False
else:
if verbose:
logger.info("Airtime API version found: " + str(api_version))
logger.info(
self.logger.info("Airtime API version found: " + str(api_version))
self.logger.info(
"pypo is only compatible with API version: " + AIRTIME_API_VERSION
)
return True
@ -140,7 +140,6 @@ class ApiClient:
return None
def upload_recorded_show(self, files, show_id):
logger = self.logger
response = ""
retries = self.UPLOAD_RETRIES
@ -148,19 +147,19 @@ class ApiClient:
url = self.construct_rest_url("upload_file_url")
logger.debug(url)
self.logger.debug(url)
for i in range(0, retries):
logger.debug("Upload attempt: %s", i + 1)
logger.debug(files)
logger.debug(ApiRequest.API_HTTP_REQUEST_TIMEOUT)
self.logger.debug("Upload attempt: %s", i + 1)
self.logger.debug(files)
self.logger.debug(ApiRequest.API_HTTP_REQUEST_TIMEOUT)
try:
request = requests.post(
url, files=files, timeout=float(ApiRequest.API_HTTP_REQUEST_TIMEOUT)
)
response = request.json()
logger.debug(response)
self.logger.debug(response)
# FIXME: We need to tell LibreTime that the uploaded track was recorded
# for a specific show
@ -184,11 +183,11 @@ class ApiClient:
break
except requests.exceptions.HTTPError as exception:
logger.error(f"Http error code: {exception.response.status_code}")
logger.exception(exception)
self.logger.error(f"Http error code: {exception.response.status_code}")
self.logger.exception(exception)
except requests.exceptions.ConnectionError as exception:
logger.exception(f"Server is down: {exception}")
self.logger.exception(f"Server is down: {exception}")
except Exception as exception:
self.logger.exception(exception)
@ -229,7 +228,6 @@ class ApiClient:
return self.services.register_component(component=component)
def notify_liquidsoap_status(self, msg, stream_id, time):
logger = self.logger
try:
# encoded_msg is no longer used server_side!!
encoded_msg = urllib.parse.quote("dummy")

View File

@ -42,8 +42,8 @@ def test_apc_url_incomplete():
def test_api_request_init():
u = ApiRequest("request_name", ApcUrl("/test/ing"))
assert u.name == "request_name"
req = ApiRequest("request_name", ApcUrl("/test/ing"))
assert req.name == "request_name"
def test_api_request_call_json():