Execute code in a notebook cell. :param code: Code to execute :param kernel_id: Kernel id to use, if not provided the default kernel will be used :param on_stdout: Callback for stdout messages :param on_stderr: Callback for stderr messages :return: Re
(
self,
code: str,
kernel_id: Optional[str] = None,
on_stdout: Optional[Callable[[ProcessMessage], Any]] = None,
on_stderr: Optional[Callable[[ProcessMessage], Any]] = None,
)
| 51 | self._start_connecting_to_default_kernel() |
| 52 | |
| 53 | def exec_cell( |
| 54 | self, |
| 55 | code: str, |
| 56 | kernel_id: Optional[str] = None, |
| 57 | on_stdout: Optional[Callable[[ProcessMessage], Any]] = None, |
| 58 | on_stderr: Optional[Callable[[ProcessMessage], Any]] = None, |
| 59 | ) -> Result: |
| 60 | """ |
| 61 | Execute code in a notebook cell. |
| 62 | :param code: Code to execute |
| 63 | :param kernel_id: Kernel id to use, if not provided the default kernel will be used |
| 64 | :param on_stdout: Callback for stdout messages |
| 65 | :param on_stderr: Callback for stderr messages |
| 66 | :return: Result of the execution |
| 67 | """ |
| 68 | kernel_id = kernel_id or self.default_kernel_id |
| 69 | ws = self._connected_kernels.get(kernel_id) |
| 70 | |
| 71 | if not ws: |
| 72 | ws = JupyterKernelWebSocket( |
| 73 | url=f"{self._sandbox.get_protocol('ws')}://{self._sandbox.get_hostname(8888)}/api/kernels/{kernel_id}/channels", |
| 74 | ) |
| 75 | self._connected_kernels[kernel_id] = ws |
| 76 | ws.connect() |
| 77 | |
| 78 | session_id = ws.send_execution_message(code, on_stdout, on_stderr) |
| 79 | |
| 80 | result = ws.get_result(session_id) |
| 81 | return result |
| 82 | |
| 83 | @property |
| 84 | def default_kernel_id(self) -> str: |