129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
import logging
|
|
import os
|
|
import pathlib
|
|
from typing import List
|
|
|
|
from money.currency import Currency
|
|
from xecd_rates_client import XecdClient
|
|
|
|
from xexe.exchange_rates import ExchangeRates
|
|
from xexe.rate_fetching import MockRateFetcher, RateFetcher, XERateFetcher
|
|
from xexe.rate_writing import CSVRateWriter, RateWriter
|
|
from xexe.utils import DateRange, generate_currency_and_dates_combinations
|
|
|
|
logger = logging.getLogger()
|
|
|
|
|
|
def run_xe_healthcheck() -> None:
|
|
"""
|
|
Try to request the account data in xe.com's API.
|
|
|
|
If certain fields about the account are returned, it means the request was
|
|
successful.
|
|
|
|
"""
|
|
logger.info("Creating client.")
|
|
xecd = XecdClient(
|
|
account_id=os.environ["XE_ACCOUNT_ID"],
|
|
api_key=os.environ["XE_API_KEY"],
|
|
)
|
|
logger.info("Requesting account info.")
|
|
|
|
try:
|
|
account_info_response = xecd.account_info()
|
|
except Exception as e:
|
|
logger.error(
|
|
"There was an exception when trying to reach xe.com. See details below."
|
|
)
|
|
logger.error(e)
|
|
raise e
|
|
|
|
contains_good_response_fields = bool(
|
|
("id" in account_info_response.keys())
|
|
and ("organization" in account_info_response.keys())
|
|
and ("package" in account_info_response.keys()),
|
|
)
|
|
if not contains_good_response_fields:
|
|
logger.error("Didn't find the fields of a good response.")
|
|
raise ConnectionError("Response from xe.com is not successful.")
|
|
|
|
logger.info("xe.com reached successfully.")
|
|
logger.info("See response below.")
|
|
logger.info(account_info_response)
|
|
|
|
|
|
def run_get_rates(
|
|
date_range: DateRange,
|
|
currencies: List[Currency],
|
|
dry_run: bool,
|
|
output: pathlib.Path,
|
|
) -> None:
|
|
logger.info("Getting rates")
|
|
|
|
process_state = GetRatesProcessState(output=output, dry_run=dry_run)
|
|
|
|
rates = obtain_rates_from_source(
|
|
process_state, date_range=date_range, currencies=currencies
|
|
)
|
|
write_rates_to_output(process_state, rates)
|
|
|
|
|
|
def obtain_rates_from_source(
|
|
process_state, date_range: DateRange, currencies: List[Currency]
|
|
) -> ExchangeRates:
|
|
rates_fetcher = process_state.get_fetcher()
|
|
|
|
currency_and_date_combinations = generate_currency_and_dates_combinations(
|
|
date_range=date_range, currencies=currencies
|
|
)
|
|
|
|
rates = ExchangeRates()
|
|
for combination in currency_and_date_combinations:
|
|
try:
|
|
rate = rates_fetcher.fetch_rate(
|
|
from_currency=combination["from_currency"],
|
|
to_currency=combination["to_currency"],
|
|
rate_date=combination["date"],
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error while fetching rates.")
|
|
logger.error(e, exc_info=True)
|
|
raise ConnectionError(f"Could not fetch rates. See logs.")
|
|
|
|
rates.add_rate(rate)
|
|
|
|
return rates
|
|
|
|
|
|
def write_rates_to_output(process_state, rates):
|
|
rates_writer = process_state.get_writer()
|
|
rates_writer.write_rates(rates)
|
|
|
|
|
|
class GetRatesProcessState:
|
|
def __init__(self, output: str, dry_run: bool) -> None:
|
|
self.writer = self._select_writer(output)
|
|
self.fetcher = self._select_fetcher(dry_run)
|
|
|
|
@staticmethod
|
|
def _select_writer(output: str) -> CSVRateWriter:
|
|
output_is_csv_file_path = bool(pathlib.Path(output).suffix == ".csv")
|
|
|
|
if output_is_csv_file_path:
|
|
return CSVRateWriter(output_file_path=output)
|
|
|
|
raise ValueError(f"Don't know how to handle passed output: {output}")
|
|
|
|
@staticmethod
|
|
def _select_fetcher(dry_run: bool) -> str:
|
|
if dry_run:
|
|
return MockRateFetcher
|
|
|
|
if not dry_run:
|
|
return XERateFetcher
|
|
|
|
def get_fetcher(self) -> RateFetcher:
|
|
return self.fetcher
|
|
|
|
def get_writer(self) -> RateWriter:
|
|
return self.writer
|