Execute Python code locally. Args: code: Python code to execute. input_spec: Input specification. Returns: Tuple of (stdout, stderr, exit_code).
(
self, code: str,
input_spec: ExecutionInput)
| 640 | return False, str(e) |
| 641 | |
| 642 | async def _local_execute_python_code( |
| 643 | self, code: str, |
| 644 | input_spec: ExecutionInput) -> tuple[str, str, int]: |
| 645 | """ |
| 646 | Execute Python code locally. |
| 647 | |
| 648 | Args: |
| 649 | code: Python code to execute. |
| 650 | input_spec: Input specification. |
| 651 | |
| 652 | Returns: |
| 653 | Tuple of (stdout, stderr, exit_code). |
| 654 | """ |
| 655 | # Install requirements first if any |
| 656 | if input_spec.requirements: |
| 657 | success, error = await self._local_install_requirements( |
| 658 | input_spec.requirements) |
| 659 | if not success: |
| 660 | return '', f'Failed to install requirements: {error}', -1 |
| 661 | |
| 662 | # Write code to temp file |
| 663 | script_file = self.scripts_dir / f'_temp_{uuid.uuid4().hex[:8]}.py' |
| 664 | try: |
| 665 | # Generate environment setup |
| 666 | env_setup = self._generate_local_env_setup(input_spec) |
| 667 | full_code = env_setup + '\n' + code |
| 668 | |
| 669 | with open(script_file, 'w', encoding='utf-8') as f: |
| 670 | f.write(full_code) |
| 671 | |
| 672 | # Build command |
| 673 | cmd = [self._get_python_executable(), str(script_file)] |
| 674 | cmd.extend([str(arg) for arg in input_spec.args]) |
| 675 | |
| 676 | # Use working_dir from input_spec for proper resource access |
| 677 | cwd = input_spec.working_dir if input_spec.working_dir else None |
| 678 | |
| 679 | stdout, stderr, exit_code = self._local_run_subprocess( |
| 680 | cmd, |
| 681 | env=input_spec.env_vars, |
| 682 | cwd=cwd, |
| 683 | stdin_input=input_spec.stdin) |
| 684 | |
| 685 | # Keep script in scripts folder for logging/debugging |
| 686 | return stdout, stderr, exit_code |
| 687 | except Exception as e: |
| 688 | logger.error(f'Local Python execution failed: {e}') |
| 689 | raise |
| 690 | |
| 691 | async def _local_execute_shell( |
| 692 | self, command: str, |
no test coverage detected