import datetime import importlib.metadata import logging import pathlib from typing import Union import click import pyfiglet from dotenv import load_dotenv from xexe.constants import PATHS, RATES_SOURCES from xexe.inputs_handling import handle_get_rates_inputs from xexe.processes import run_dwh_healthcheck, run_get_rates, run_xe_healthcheck logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - [%(levelname)s] - %(filename)s - L%(lineno)d - %(message)s", handlers=[logging.FileHandler(PATHS.logging_file), logging.StreamHandler()], ) logger = logging.getLogger() load_dotenv(dotenv_path=PATHS.dot_env_file_path, verbose=True) @click.group() def cli(): logger.info(pyfiglet.figlet_format("\nWelcome to xexe", font="big")) logger.info(f"Running xexe version: {importlib.metadata.version('xexe')}") @cli.command() def smoke_test(): print("Oink oink!") print( """ __,---.__ ,-' `-.__ &/ `._\ _\\ / ''._ | , (") |__,'`-..--|__|--'' """ ) @cli.command() def xe_healthcheck(): logger.info("Running healthcheck against xe.com API.") run_xe_healthcheck() logger.info("Healthcheck attempt finished.") @cli.command() def dwh_healthcheck(): logger.info("Running healthcheck against dwh API.") run_dwh_healthcheck() logger.info("Healthcheck attempt finished.") @cli.command() @click.option( "--start-date", type=click.DateTime(), default=datetime.datetime.today() - datetime.timedelta(days=7), ) @click.option( "--end-date", type=click.DateTime(), default=datetime.datetime.today(), ) @click.option( "--currencies", default=",".join([]), show_default=True, type=click.STRING ) @click.option("--pairs", default=",".join([]), show_default=True, type=click.STRING) @click.option("--dry-run", is_flag=True) @click.option("--rates-source", type=click.Choice(RATES_SOURCES.keys()), default="mock") @click.option("--ignore-warnings", is_flag=True) @click.option("--output", type=click.STRING, required=True) def get_rates( start_date: Union[str, datetime.datetime, datetime.date], end_date: Union[str, datetime.datetime, datetime.date], currencies: Union[None, str], pairs: Union[None, str], dry_run: bool, rates_source: str, ignore_warnings: bool, output: pathlib.Path, ): get_rates_inputs = handle_get_rates_inputs( start_date=start_date, end_date=end_date, currencies=currencies, pairs=pairs, dry_run=dry_run, rates_source=rates_source, ignore_warnings=ignore_warnings, output=output, ) logger.info("Starting get-rates process.") run_get_rates(**get_rates_inputs) logger.info("Finished get-rates process.")