from pathlib import Path import logging import click from iottb.definitions import DB_NAME, CFG_FILE_PATH from iottb.models.iottb_config import IottbConfig logger = logging.getLogger(__name__) @click.group('util') def tb(): pass @click.command() @click.option('--file', default=DB_NAME) @click.option('--table', type=str, default='DefaultDatabase') @click.option('--key') @click.option('--value') @click.pass_context def set_key_in_table_to(ctx, file, table, key, value): """Edit config or metadata files. TODO: Implement""" click.echo(f'set_key_in_table_to invoked') logger.warning("Unimplemented subcommand invoked.") @click.command() @click.confirmation_option(prompt="Are you certain that you want to delete the cfg file?") def rm_cfg(): """ Removes the cfg file from the filesystem. This is mostly a utility during development. Once non-standard database locations are implemented, deleting this would lead to iottb not being able to find them anymore. """ Path(CFG_FILE_PATH).unlink() click.echo(f'Iottb configuration removed at {CFG_FILE_PATH}') @click.command() @click.confirmation_option(prompt="Are you certain that you want to delete the databases file?") def rm_dbs(dbs): """ Removes ALL(!) databases from the filesystem if they're empty. Development utility currently unfit for use. """ config = IottbConfig() paths = config.get_know_database_paths() logger.debug(f'Known db paths: {str(paths)}') for dbs in paths: try: Path(dbs).rmdir() click.echo(f'{dbs} deleted') except Exception as e: logger.debug(f'Failed unlinking db {dbs} with error {e}') logger.info(f'All databases deleted') @click.command('show-cfg', help='Show the current configuration context') @click.option('--cfg-file', type=click.Path(), default=CFG_FILE_PATH, help='Path to the config file') @click.option('-pp', is_flag=True, default=False, help='Pretty Print') @click.pass_context def show_cfg(ctx, cfg_file, pp): logger.debug(f'Pretty print option set to {pp}') if pp: try: config = IottbConfig(Path(cfg_file)) click.echo("Configuration Context:") click.echo(f"Default Database: {config.default_database}") click.echo(f"Default Database Path: {config.default_db_location}") click.echo("Database Locations:") for db_name, db_path in config.db_path_dict.items(): click.echo(f" - {db_name}: {db_path}") except Exception as e: logger.error(f"Error loading configuration: {e}") click.echo(f"Failed to load configuration from {cfg_file}") else: path = Path(cfg_file) if path.is_file(): with path.open('r') as file: content = file.read() click.echo(content) else: click.echo(f"Configuration file not found at {cfg_file}") @click.command('show-all', help='Show everything: configuration, databases, and device metadata') @click.pass_context def show_everything(ctx): """Show everything that can be recursively found based on config except file contents.""" config = ctx.obj['CONFIG'] click.echo("Configuration Context:") click.echo(f"Default Database: {config.default_database}") click.echo(f"Default Database Path: {config.default_db_location}") click.echo("Database Locations:") for db_name, db_path in config.db_path_dict.items(): full_db_path = Path(db_path) / db_name click.echo(f" - {db_name}: {full_db_path}") if full_db_path.is_dir(): click.echo(f"Contents of {db_name} at {full_db_path}:") for item in full_db_path.iterdir(): if item.is_file(): click.echo(f" - {item.name}") try: with item.open('r', encoding='utf-8') as file: content = file.read() click.echo(f" Content:\n{content}") except UnicodeDecodeError: click.echo(" Content is not readable as text") elif item.is_dir(): click.echo(f" - {item.name}/") for subitem in item.iterdir(): if subitem.is_file(): click.echo(f" - {subitem.name}") elif subitem.is_dir(): click.echo(f" - {subitem.name}/") else: click.echo(f" {full_db_path} is not a directory") warnstyle = {'fg': 'red', 'bold': True} click.secho('Developer command used', **warnstyle)