(
self,
sandbox_id: str,
command: Sequence[str],
*,
stream_output: bool = False,
workdir: str | None = None,
env: Mapping[str, str] | None = None,
stdin: bytes | None = None,
timeout_seconds: int | None = None,
)
| 525 | ) |
| 526 | |
| 527 | def exec( |
| 528 | self, |
| 529 | sandbox_id: str, |
| 530 | command: Sequence[str], |
| 531 | *, |
| 532 | stream_output: bool = False, |
| 533 | workdir: str | None = None, |
| 534 | env: Mapping[str, str] | None = None, |
| 535 | stdin: bytes | None = None, |
| 536 | timeout_seconds: int | None = None, |
| 537 | ) -> ExecResult: |
| 538 | result: ExecResult | None = None |
| 539 | for item in self.exec_stream( |
| 540 | sandbox_id, |
| 541 | command, |
| 542 | workdir=workdir, |
| 543 | env=env, |
| 544 | stdin=stdin, |
| 545 | timeout_seconds=timeout_seconds, |
| 546 | ): |
| 547 | if stream_output and isinstance(item, ExecChunk): |
| 548 | if item.stream == "stdout": |
| 549 | sys.stdout.buffer.write(item.data) |
| 550 | sys.stdout.flush() |
| 551 | else: |
| 552 | sys.stderr.buffer.write(item.data) |
| 553 | sys.stderr.flush() |
| 554 | if isinstance(item, ExecResult): |
| 555 | result = item |
| 556 | if result is None: |
| 557 | raise SandboxError("ExecSandbox did not return a result") |
| 558 | return result |
| 559 | |
| 560 | def exec_python( |
| 561 | self, |
no test coverage detected