(mycli: 'MyCli', cli_args: 'CliArgs')
| 186 | |
| 187 | |
| 188 | def main_batch_without_progress_bar(mycli: 'MyCli', cli_args: 'CliArgs') -> int: |
| 189 | if cli_args.batch is None: |
| 190 | return 1 |
| 191 | if not sys.stdin.isatty() and cli_args.batch != '-': |
| 192 | click.secho('Ignoring STDIN since --batch was also given.', err=True, fg='red') |
| 193 | try: |
| 194 | completed_statement_count = replay_checkpoint_file(cli_args.batch, cli_args.checkpoint, cli_args.resume) |
| 195 | batch_h = click.open_file(cli_args.batch) |
| 196 | except (OSError, FileNotFoundError): |
| 197 | click.secho(f'Failed to open --batch file: {cli_args.batch}', err=True, fg='red') |
| 198 | return 1 |
| 199 | except CheckpointReplayError as e: |
| 200 | name = cli_args.checkpoint if cli_args.checkpoint else 'None' |
| 201 | click.secho(f'Error replaying --checkpoint file: {name}: {e}', err=True, fg='red') |
| 202 | return 1 |
| 203 | try: |
| 204 | for statement, counter in statements_from_filehandle(batch_h): |
| 205 | if counter < completed_statement_count: |
| 206 | continue |
| 207 | dispatch_batch_statements(mycli, cli_args, statement, counter) |
| 208 | except (ValueError, StopIteration, IOError, OSError, pymysql.err.Error) as e: |
| 209 | click.secho(str(e), err=True, fg='red') |
| 210 | return 1 |
| 211 | finally: |
| 212 | batch_h.close() |
| 213 | return 0 |
| 214 | |
| 215 | |
| 216 | def main_batch_from_stdin(mycli: 'MyCli', cli_args: 'CliArgs') -> int: |
no test coverage detected