Files
@ 1927a1812033
Branch filter:
Location: NPO-Accounting/oxrlib/oxrlib/loaders.py - annotation
1927a1812033
3.2 KiB
text/x-python
loaders: Add LoaderChain.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 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 | 10b0a818d759 1927a1812033 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 744479013d7a 1927a1812033 1927a1812033 1927a1812033 1927a1812033 d1ba2dbc6a33 d1ba2dbc6a33 d1ba2dbc6a33 d1ba2dbc6a33 d1ba2dbc6a33 d1ba2dbc6a33 d1ba2dbc6a33 744479013d7a 744479013d7a 744479013d7a 744479013d7a 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 10b0a818d759 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 1927a1812033 | import cgi
import functools
import io
import urllib.request
import urllib.parse
class LoaderError(Exception):
pass
class LoaderNoDataError(LoaderError):
pass
class LoaderBadRequestError(LoaderError):
pass
class LoaderSourceError(LoaderError):
pass
class NoLoadersError(Exception):
pass
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
class OXRAPIRequest:
DEFAULT_API_ROOT = 'https://openexchangerates.org/api/'
DEFAULT_RESPONSE_ENCODING = 'utf-8'
def __init__(self, app_id, api_root=None, *, open_func=urllib.request.urlopen):
self.api_root = self.DEFAULT_API_ROOT if api_root is None else api_root
self.app_id = app_id
self.open_url = open_func
def _get_response_encoding(self, response, default=None):
try:
content_type = response.getheader('Content-Type', 'application/json')
_, ct_options = cgi.parse_header(content_type)
encoding = ct_options['charset']
except (KeyError, ValueError):
encoding = self.DEFAULT_RESPONSE_ENCODING if default is None else default
return encoding
def _raw_query(self, url_tail, params):
url = '{}?{}'.format(
urllib.parse.urljoin(self.api_root, url_tail),
urllib.parse.urlencode(params),
)
response = self.open_url(url)
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
elif status_code >= 500:
exc_class = LoaderSourceError
else:
exc_class = LoaderBadRequestError
with response_body:
raise exc_class(url, response_body.read(64 * 1024))
def historical(self, date, base):
return self._raw_query(
'historical/{}.json'.format(date.isoformat()),
{'app_id': self.app_id, 'base': base},
)
class LoaderChain:
def __init__(self):
self.loaders = []
def add_loader(self, loader):
self.loaders.append(loader)
def _wrap_load_method(orig_func):
@functools.wraps(orig_func)
def load_wrapper(self, *args, **kwargs):
self.used_loader = None
error = None
for loader in self.loaders:
try:
response = getattr(loader, orig_func.__name__)(*args, **kwargs)
except LoaderError as this_error:
error = this_error
else:
self.used_loader = loader
return response
else:
raise NoLoadersError() if error is None else error
return load_wrapper
@_wrap_load_method
def historical(self, date, base):
pass
|