import cgi
import functools
import io
import urllib.request
import urllib.parse
class LoaderError(Exception):
pass
class LoaderNoDataError(LoaderError):
class LoaderBadRequestError(LoaderError):
class LoaderSourceError(LoaderError):
class NoLoadersError(Exception):
from . import errors
class FileCache:
def __init__(self, dir_path, filename_pattern):
self.dir_path = dir_path
self.pattern = filename_pattern
def historical(self, date, base):
path = self.dir_path / self.pattern.format(date=date.isoformat(), base=base)
try:
return path.open()
except FileNotFoundError as error:
raise LoaderNoDataError(path) from error
raise errors.LoaderNoDataError(path) from error
class OXRAPIRequest:
DEFAULT_API_ROOT = 'https://openexchangerates.org/api/'
DEFAULT_RESPONSE_ENCODING = 'utf-8'
@@ -64,17 +46,17 @@ class OXRAPIRequest:
status_code = response.status
encoding = self._get_response_encoding(response)
response_body = io.TextIOWrapper(response, encoding=encoding)
if 200 <= status_code < 203:
return response_body
elif status_code == 404 or status_code == 410:
exc_class = LoaderNoDataError
exc_class = errors.LoaderNoDataError
elif status_code >= 500:
exc_class = LoaderSourceError
exc_class = errors.LoaderSourceError
else:
exc_class = LoaderBadRequestError
exc_class = errors.LoaderBadRequestError
with response_body:
raise exc_class(url, response_body.read(64 * 1024))
return self._raw_query(
'historical/{}.json'.format(date.isoformat()),
@@ -94,18 +76,18 @@ class LoaderChain:
def load_wrapper(self, *args, **kwargs):
self.used_loader = None
error = None
for loader in self.loaders:
response = getattr(loader, orig_func.__name__)(*args, **kwargs)
except LoaderError as this_error:
except errors.LoaderError as this_error:
error = this_error
self.used_loader = loader
return response
raise NoLoadersError() if error is None else error
raise errors.NoLoadersError() if error is None else error
return load_wrapper
@_wrap_load_method