99 lines
2.7 KiB
Python
99 lines
2.7 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.utils import DateRange
|
|
|
|
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)
|
|
|
|
rates_fetcher = get_rates_fetcher(process_state.fetcher_type)
|
|
|
|
try:
|
|
rates = rates_fetcher.fetch()
|
|
except Exception as e:
|
|
process_state.record_rate_fetching_error(e)
|
|
raise ConnectionError(f"Could not fetch rates. See logs.")
|
|
|
|
rates_writer = get_rates_writer(process_state.output_type)
|
|
|
|
try:
|
|
rates_writer.write(rates)
|
|
except Exception as e:
|
|
process_state.record_rate_writing_error(e)
|
|
raise Exception(f"Could not write rates. See logs.")
|
|
|
|
|
|
class GetRatesProcessState:
|
|
def __init__(self, output: str, dry_run: bool) -> None:
|
|
self.writer_type = self._infer_writer_type(output)
|
|
self.fetcher_type = self._infer_fetcher_type(dry_run)
|
|
|
|
@staticmethod
|
|
def _infer_writer_type(output: str) -> str:
|
|
output_is_csv_file_path = bool(pathlib.Path(output).suffix == ".csv")
|
|
|
|
if output_is_csv_file_path:
|
|
return "csv_file"
|
|
|
|
raise ValueError(f"Don't know how to handle passed output: {output}")
|
|
|
|
@staticmethod
|
|
def _infer_fetcher_type(dry_run: bool) -> str:
|
|
if dry_run:
|
|
return MockFetcher
|
|
|
|
if not dry_run:
|
|
return XEFetcher
|