import datetime import logging import pathlib from typing import Union from money.currency import Currency from xexe.constants import DEFAULT_CURRENCIES, RATES_SOURCES from xexe.currency_pair import CurrencyPair from xexe.utils import DateRange logger = logging.getLogger() def handle_get_rates_inputs( start_date: Union[datetime.datetime, datetime.date], end_date: Union[datetime.datetime, datetime.date], dry_run: bool, rates_source: str, ignore_warnings: bool, output: Union[str, pathlib.Path], currencies: Union[None, str] = None, pairs: Union[None, str] = None, ): logger.info("Handling inputs.") date_range = DateRange(start_date=start_date.date(), end_date=end_date.date()) if date_range.end_date > datetime.datetime.today().date(): date_range.end_date = datetime.datetime.today().date() if pairs: if currencies: logger.error(f"Received both currencies and pairs.") logger.error(f"Currencies: '{currencies}'.") logger.error(f"Pairs: '{pairs}'.") raise ValueError("You can pass currencies or pairs, but not both.") pairs = { CurrencyPair( from_currency=Currency[str_pair[0:3]], to_currency=Currency[str_pair[3:6]], ) for str_pair in pairs.split(",") } if currencies: # CLI input comes as a string of comma-separated currency codes currencies = {currency_code.strip() for currency_code in currencies.split(",")} tmp = {Currency(currency_code) for currency_code in currencies} currencies = tmp if currencies is None or currencies == "" and not pairs: logger.info("No currency list or pairs passed. Running for default currencies.") currencies = DEFAULT_CURRENCIES if rates_source not in RATES_SOURCES: raise ValueError(f"--rates-source must be one of {RATES_SOURCES.keys()}.") if not output == "dwh": # The Path constructor is idempotent, so this works equally fine if output # is a string or an actual Path object. output = pathlib.Path(output) if output.suffix != ".csv": raise ValueError("File output must be a .csv file.") prepared_inputs = { "date_range": date_range, "dry_run": dry_run, "rates_source": rates_source, "ignore_warnings": ignore_warnings, "output": output, } if currencies: prepared_inputs["currencies"] = currencies if pairs: prepared_inputs["pairs"] = pairs logger.debug(prepared_inputs) return prepared_inputs