/
base.py
83 lines (67 loc) · 2.29 KB
/
base.py
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
import json
import ssl
from decimal import Decimal
from urllib.parse import parse_qsl, urlparse, urlunparse
from urllib.request import urlopen
from django.db.transaction import atomic
from django.utils.http import urlencode
from djmoney import settings
from ..models import ExchangeBackend, Rate
try:
import certifi
except ImportError:
raise ImportError("Please install dependency certifi - pip install certifi")
class BaseExchangeBackend:
name = None
url = None
def get_rates(self, **kwargs):
"""
Returns a mapping <currency>: <rate>.
"""
raise NotImplementedError
def get_url(self, **params):
"""
Updates base url with provided GET parameters.
"""
parts = list(urlparse(self.url))
query = dict(parse_qsl(parts[4]))
query.update(params)
parts[4] = urlencode(query)
return urlunparse(parts)
def get_params(self):
"""
Default GET parameters for the request.
"""
return {}
def get_response(self, **params):
url = self.get_url(**params)
context = ssl.create_default_context(cafile=certifi.where())
response = urlopen(url, context=context)
return response.read()
def parse_json(self, response):
if isinstance(response, bytes):
response = response.decode("utf-8")
return json.loads(response, parse_float=Decimal)
@atomic
def update_rates(self, base_currency=settings.BASE_CURRENCY, **kwargs):
"""
Updates rates for the given backend.
"""
backend, _ = ExchangeBackend.objects.update_or_create(name=self.name, defaults={"base_currency": base_currency})
backend.clear_rates()
params = self.get_params()
params.update(base_currency=base_currency, **kwargs)
Rate.objects.bulk_create(
[
Rate(currency=currency, value=value, backend=backend)
for currency, value in self.get_rates(**params).items()
]
)
class SimpleExchangeBackend(BaseExchangeBackend):
"""
Simple backend implementation.
Assumes JSON response with `rates` key inside.
"""
def get_rates(self, **params):
response = self.get_response(**params)
return self.parse_json(response)["rates"]