(
mycli: 'MyCli',
cli_args: 'CliArgs',
statements: str,
batch_counter: int,
)
| 90 | |
| 91 | |
| 92 | def dispatch_batch_statements( |
| 93 | mycli: 'MyCli', |
| 94 | cli_args: 'CliArgs', |
| 95 | statements: str, |
| 96 | batch_counter: int, |
| 97 | ) -> None: |
| 98 | if batch_counter: |
| 99 | if cli_args.format == 'csv': |
| 100 | mycli.main_formatter.format_name = 'csv-noheader' |
| 101 | elif cli_args.format == 'tsv': |
| 102 | mycli.main_formatter.format_name = 'tsv_noheader' |
| 103 | elif cli_args.format == 'table': |
| 104 | mycli.main_formatter.format_name = 'ascii' |
| 105 | else: |
| 106 | mycli.main_formatter.format_name = 'tsv' |
| 107 | else: |
| 108 | if cli_args.format == 'csv': |
| 109 | mycli.main_formatter.format_name = 'csv' |
| 110 | elif cli_args.format == 'tsv': |
| 111 | mycli.main_formatter.format_name = 'tsv' |
| 112 | elif cli_args.format == 'table': |
| 113 | mycli.main_formatter.format_name = 'ascii' |
| 114 | else: |
| 115 | mycli.main_formatter.format_name = 'tsv' |
| 116 | |
| 117 | execution_confirmed: bool | None = True |
| 118 | if cli_args.warn_batch and is_destructive(mycli.destructive_keywords, statements): |
| 119 | try: |
| 120 | # this seems to work, even though we are reading from stdin above |
| 121 | sys.stdin = open('/dev/tty') |
| 122 | execution_confirmed = confirm_destructive_query(mycli.destructive_keywords, statements) |
| 123 | except (IOError, OSError) as e: |
| 124 | mycli.logger.warning('Unable to open TTY as stdin.') |
| 125 | raise e |
| 126 | if execution_confirmed: |
| 127 | if cli_args.throttle > 0 and batch_counter >= 1: |
| 128 | time.sleep(cli_args.throttle) |
| 129 | mycli.run_query(statements, checkpoint=cli_args.checkpoint, new_line=True) |
| 130 | |
| 131 | |
| 132 | def main_batch_with_progress_bar(mycli: 'MyCli', cli_args: 'CliArgs') -> int: |
no test coverage detected