class_docstr: The IPython Kernel class.
class_docstr_tok: ["The","IPython","Kernel","class","."]
docstr: ContextManager for capturing SIGINT and cancelling a future
  
  SIGINT raises in the event loop when running async code,
  but we want it to halt a coroutine.
  
  Ideally, it would raise KeyboardInterrupt,
  but this turns it into a CancelledError.
  At least it gets a decent traceback to the user.
docstr_tok: ["ContextManager","for","capturing","SIGINT","and","cancelling","a","future","SIGINT","raises","in","the","event","loop","when","running","async","code",",","but","we","want","it","to","halt","a","coroutine",".","Ideally",",","it","would","raise","KeyboardInterrupt",",","but","this","turns","it","into","a","CancelledError",".","At","least","it","gets","a","decent","traceback","to","the","user","."]
code: def _cancel_on_sigint(self, future):
      """ContextManager for capturing SIGINT and cancelling a future
  
      SIGINT raises in the event loop when running async code,
      but we want it to halt a coroutine.
  
      Ideally, it would raise KeyboardInterrupt,
      but this turns it into a CancelledError.
      At least it gets a decent traceback to the user.
      """
      sigint_future: asyncio.Future[int] = asyncio.Future()
  
      # whichever future finishes first,
      # cancel the other one
      def cancel_unless_done(f, _ignored):
          if f.cancelled() or f.done():
              return
          f.cancel()
  
      # when sigint finishes,
      # abort the coroutine with CancelledError
      sigint_future.add_done_callback(
          partial(cancel_unless_done, future)
      )
      # when the main future finishes,
      # stop watching for SIGINT events
      future.add_done_callback(
          partial(cancel_unless_done, sigint_future)
      )
  
      def handle_sigint(*args):
          def set_sigint_result():
              if sigint_future.cancelled() or sigint_future.done():
                  return
              sigint_future.set_result(1)
  
          # use add_callback for thread safety
          self.io_loop.add_callback(set_sigint_result)
  
      # set the custom sigint handler during this context
      save_sigint = signal.signal(signal.SIGINT, handle_sigint)
      try:
          yield
      finally:
          # restore the previous sigint handler
          signal.signal(signal.SIGINT, save_sigint)
  
imports: import asyncio
  import builtins
  import getpass
  import os
  import signal
  import sys
  import threading
  import typing
  from contextlib import contextmanager
  from functools import partial
  import comm
  from IPython.core import release
  from IPython.utils.tokenutil import line_at_cursor, token_at_cursor
  from traitlets import Any, Bool, HasTraits, Instance, List, Type, observe, observe_compat
  from zmq.eventloop.zmqstream import ZMQStream
  from .comm.comm import BaseComm
  from .comm.manager import CommManager
  from .compiler import XCachingCompiler
  from .debugger import Debugger, _is_debugpy_available
  from .eventloops import _use_appnope
  from .kernelbase import Kernel
  from .kernelbase import _accepts_parameters
  from .zmqshell import ZMQInteractiveShell
label_desc: Use image node_id 14 for calling the IPythonKernel obj's underlying member method code with example usage: obj._cancel_on_sigint(future) without return types