| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114 |
- """A Jupyter console app to run files."""
- # Copyright (c) Jupyter Development Team.
- # Distributed under the terms of the Modified BSD License.
- from __future__ import annotations
- import atexit
- import signal
- import sys
- import typing as t
- from jupyter_core.application import JupyterApp, base_aliases, base_flags
- from traitlets import Any, Dict, Float
- from traitlets.config import catch_config_error
- from . import __version__
- from .consoleapp import JupyterConsoleApp, app_aliases, app_flags
- OUTPUT_TIMEOUT = 10
- # copy flags from mixin:
- flags = dict(base_flags)
- # start with mixin frontend flags:
- frontend_flags_dict = dict(app_flags)
- # update full dict with frontend flags:
- flags.update(frontend_flags_dict)
- # copy flags from mixin
- aliases = dict(base_aliases)
- # start with mixin frontend flags
- frontend_aliases_dict = dict(app_aliases)
- # load updated frontend flags into full dict
- aliases.update(frontend_aliases_dict)
- # get flags&aliases into sets, and remove a couple that
- # shouldn't be scrubbed from backend flags:
- frontend_aliases = set(frontend_aliases_dict.keys())
- frontend_flags = set(frontend_flags_dict.keys())
- class RunApp(JupyterApp, JupyterConsoleApp):
- """An Jupyter Console app to run files."""
- version = __version__
- name = "jupyter run"
- description = """Run Jupyter kernel code."""
- flags = Dict(flags)
- aliases = Dict(aliases)
- frontend_aliases = Any(frontend_aliases)
- frontend_flags = Any(frontend_flags)
- kernel_timeout = Float(
- 60,
- config=True,
- help="""Timeout for giving up on a kernel (in seconds).
- On first connect and restart, the console tests whether the
- kernel is running and responsive by sending kernel_info_requests.
- This sets the timeout in seconds for how long the kernel can take
- before being presumed dead.
- """,
- )
- def parse_command_line(self, argv: list[str] | None = None) -> None:
- """Parse the command line arguments."""
- super().parse_command_line(argv)
- self.build_kernel_argv(self.extra_args)
- self.filenames_to_run = self.extra_args[:]
- @catch_config_error
- def initialize(self, argv: list[str] | None = None) -> None: # type:ignore[override]
- """Initialize the app."""
- self.log.debug("jupyter run: initialize...")
- super().initialize(argv)
- JupyterConsoleApp.initialize(self)
- signal.signal(signal.SIGINT, self.handle_sigint)
- if self.kernel_manager:
- atexit.register(self.kernel_manager.shutdown_kernel)
- def handle_sigint(self, *args: t.Any) -> None:
- """Handle SIGINT."""
- if self.kernel_manager:
- self.kernel_manager.interrupt_kernel()
- else:
- self.log.error("Cannot interrupt kernels we didn't start.\n")
- def start(self) -> None:
- """Start the application."""
- self.log.debug("jupyter run: starting...")
- super().start()
- self.kernel_client.wait_for_ready(timeout=self.kernel_timeout)
- if self.filenames_to_run:
- for filename in self.filenames_to_run:
- self.log.debug("jupyter run: executing `%s`", filename)
- with open(filename) as fp:
- code = fp.read()
- reply = self.kernel_client.execute_interactive(code, timeout=OUTPUT_TIMEOUT)
- return_code = 0 if reply["content"]["status"] == "ok" else 1
- if return_code:
- msg = f"jupyter-run error running '{filename}'"
- raise Exception(msg)
- else:
- self.log.debug("jupyter run: executing from stdin")
- code = sys.stdin.read()
- reply = self.kernel_client.execute_interactive(code, timeout=OUTPUT_TIMEOUT)
- return_code = 0 if reply["content"]["status"] == "ok" else 1
- if return_code:
- msg = "jupyter-run error running 'stdin'"
- raise Exception(msg)
- main = launch_new_instance = RunApp.launch_instance
- if __name__ == "__main__":
- main()
|
