81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
import datetime
|
|
import logging
|
|
import pathlib
|
|
from itertools import combinations
|
|
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 (currencies is None or currencies == "") and not pairs:
|
|
logger.info("No currency list or pairs passed. Running for default currencies.")
|
|
currencies = DEFAULT_CURRENCIES
|
|
|
|
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(",")}
|
|
currencies = {Currency(currency_code) for currency_code in currencies}
|
|
pairs = list(combinations(currencies, 2))
|
|
pairs = {
|
|
CurrencyPair(from_currency=pair[0], to_currency=pair[1]) for pair in pairs
|
|
}
|
|
|
|
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,
|
|
"pairs": pairs,
|
|
"rates_source": rates_source,
|
|
"ignore_warnings": ignore_warnings,
|
|
"output": output,
|
|
}
|
|
|
|
logger.debug(prepared_inputs)
|
|
return prepared_inputs
|