(
self, sql: str, label: str, output: str = "csv", echo_stdout: bool = True
)
| 28 | self.dry_run = dry_run |
| 29 | |
| 30 | def run( |
| 31 | self, sql: str, label: str, output: str = "csv", echo_stdout: bool = True |
| 32 | ) -> QueryResult: |
| 33 | cmd = [self.binary, "--output", output, f"--query={sql}"] |
| 34 | if self.dsn: |
| 35 | cmd[1:1] = ["--dsn", self.dsn] |
| 36 | |
| 37 | printable = " ".join(shlex.quote(part) for part in cmd) |
| 38 | print(f"\n[{label}] {printable}", flush=True) |
| 39 | start = time.monotonic() |
| 40 | if self.dry_run: |
| 41 | return QueryResult(label, sql, 0.0, "", "") |
| 42 | |
| 43 | proc = subprocess.run(cmd, text=True, capture_output=True) |
| 44 | elapsed = time.monotonic() - start |
| 45 | if proc.returncode != 0: |
| 46 | print(proc.stdout, end="", file=sys.stdout) |
| 47 | print(proc.stderr, end="", file=sys.stderr) |
| 48 | raise RuntimeError(f"bendsql failed for {label}, exit={proc.returncode}") |
| 49 | if echo_stdout and proc.stdout.strip(): |
| 50 | print(proc.stdout.strip(), flush=True) |
| 51 | if proc.stderr.strip(): |
| 52 | print(proc.stderr.strip(), file=sys.stderr, flush=True) |
| 53 | return QueryResult(label, sql, elapsed, proc.stdout, proc.stderr) |
| 54 | |
| 55 | def try_run( |
| 56 | self, sql: str, label: str, output: str = "csv", echo_stdout: bool = True |
no test coverage detected