This commit is contained in:
gr4viton 2025-01-28 16:09:14 +01:00 committed by GitHub
commit 8b5df5d336
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -1,10 +1,17 @@
import csv
import traceback
import warnings
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Dict, Optional
from cachetools import TTLCache, cached
import requests
NO_CURRENCY = "XXX"
ExchangeRates = Dict[str, float]
class Singleton(type):
_instances = {}
@ -14,205 +21,93 @@ class Singleton(type):
return cls._instances[cls]
class CurrencyConverter(object, metaclass=Singleton):
# Get exchange rates
no_currency = "XXX"
class ExchangeRateGetter(ABC):
def get_rates(self) -> Optional[ExchangeRates]:
"""Method to retrieve a list of currency conversion rates.
Returns:
currencies: dict - key is a three-letter currency, value is float of conversion to base currency
"""
try:
return self._get_rates()
except Exception:
warnings.warn(
f"Exchange rate getter failed - {traceback.format_exc(limit=0).strip()}"
)
@abstractmethod
def _get_rates(self) -> Optional[ExchangeRates]:
"""Actual implementation of the exchange rate getter."""
raise NotImplementedError
class ApiExchangeRate(ExchangeRateGetter):
api_url = "https://api.exchangerate.host/latest?base=USD"
def _get_rates(self) -> Optional[ExchangeRates]:
return requests.get(self.api_url).json()["rates"] # TODO not working currently probably
class UserExchangeRate(ExchangeRateGetter):
user_csv_file = "path/to/file.csv"
def _get_rates(self) -> Optional[ExchangeRates]:
"""Get rates from user defined csv.
The user_csv_file should contain the currency conversions to "USD" without 1 header row
Example:
```
currency_code,fx_rate_to_USD
CZK,25.0
...
```
TODO: make it work bi-directionally
TODO: document for the user where to place the file
"""
reader = csv.reader(self.user_csv_file)
rates = {}
for row in reader:
from_currency = row[0]
rate = float(row[1])
# TODO add validation and exception handling for typos
rates[from_currency] = rate
return rates
class HardCodedExchangeRate(ExchangeRateGetter):
def _get_rates(self) -> Optional[dict]:
return {"USD": 1.0} # TODO fill in more
class CurrencyConverter(object, metaclass=Singleton):
no_currency = NO_CURRENCY
def __init__(self):
pass
@cached(cache=TTLCache(maxsize=1, ttl=86400))
def get_rates(self):
try:
rates = requests.get(self.api_url).json()["rates"]
except Exception:
warnings.warn(
f"Call to {self.api_url} failed: {traceback.format_exc(limit=0).strip()}"
)
# In case of any exception, let's have an empty value
"""Try to retrieve the exchange rate from various sources, defaulting to hard coded values."""
for provider in [ApiExchangeRate, UserExchangeRate, HardCodedExchangeRate]:
if rates:= provider.get_rates():
break
else:
rates = {}
rates[self.no_currency] = 1.0
rates[NO_CURRENCY] = 1.0
return rates
def get_currencies(self, with_no_currency=True):
currencies = [
"AED",
"AFN",
"ALL",
"AMD",
"ANG",
"AOA",
"ARS",
"AUD",
"AWG",
"AZN",
"BAM",
"BBD",
"BDT",
"BGN",
"BHD",
"BIF",
"BMD",
"BND",
"BOB",
"BRL",
"BSD",
"BTC",
"BTN",
"BWP",
"BYN",
"BZD",
"CAD",
"CDF",
"CHF",
"CLF",
"CLP",
"CNH",
"CNY",
"COP",
"CRC",
"CUC",
"CUP",
"CVE",
"CZK",
"DJF",
"DKK",
"DOP",
"DZD",
"EGP",
"ERN",
"ETB",
"EUR",
"FJD",
"FKP",
"GBP",
"GEL",
"GGP",
"GHS",
"GIP",
"GMD",
"GNF",
"GTQ",
"GYD",
"HKD",
"HNL",
"HRK",
"HTG",
"HUF",
"IDR",
"ILS",
"IMP",
"INR",
"IQD",
"IRR",
"ISK",
"JEP",
"JMD",
"JOD",
"JPY",
"KES",
"KGS",
"KHR",
"KMF",
"KPW",
"KRW",
"KWD",
"KYD",
"KZT",
"LAK",
"LBP",
"LKR",
"LRD",
"LSL",
"LYD",
"MAD",
"MDL",
"MGA",
"MKD",
"MMK",
"MNT",
"MOP",
"MRU",
"MUR",
"MVR",
"MWK",
"MXN",
"MYR",
"MZN",
"NAD",
"NGN",
"NIO",
"NOK",
"NPR",
"NZD",
"OMR",
"PAB",
"PEN",
"PGK",
"PHP",
"PKR",
"PLN",
"PYG",
"QAR",
"RON",
"RSD",
"RUB",
"RWF",
"SAR",
"SBD",
"SCR",
"SDG",
"SEK",
"SGD",
"SHP",
"SLL",
"SOS",
"SRD",
"SSP",
"STD",
"STN",
"SVC",
"SYP",
"SZL",
"THB",
"TJS",
"TMT",
"TND",
"TOP",
"TRY",
"TTD",
"TWD",
"TZS",
"UAH",
"UGX",
"USD",
"UYU",
"UZS",
"VEF",
"VES",
"VND",
"VUV",
"WST",
"XAF",
"XAG",
"XAU",
"XCD",
"XDR",
"XOF",
"XPD",
"XPF",
"XPT",
"YER",
"ZAR",
"ZMW",
"ZWL",
]
def get_currencies(self, with_no_currency: bool=True) -> list:
currencies = list(self.get_rates.keys())
if with_no_currency:
currencies.append(self.no_currency)
return currencies
def exchange_currency(self, amount, source_currency, dest_currency):
def exchange_currency(self, amount: float, source_currency: str, dest_currency: str) -> float:
"""Return the money amount converted from source_currency to dest_currency."""
if (
source_currency == dest_currency
or source_currency == self.no_currency
@ -223,6 +118,8 @@ class CurrencyConverter(object, metaclass=Singleton):
rates = self.get_rates()
source_rate = rates[source_currency]
dest_rate = rates[dest_currency]
new_amount = (float(amount) / source_rate) * dest_rate
# round to two digits because we are dealing with money
return round(new_amount, 2)
# Using Decimal to not introduce floating-point operation absolute errors
new_amount = (Decimal(amount) / Decimal(source_rate)) * Decimal(dest_rate)
# dealing with money - only round the shown amount before showing it to user
# - think about 10 * 0.0003 == 0?
return float(new_amount)