Method shebang

IPython/core/magics/script.py:176–348

Run a cell via a shell command The `%%script` line is like the #! line of script, specifying a program (bash, perl, ruby, etc.) with which to run. The rest of the cell is run by that program. .. versionchanged:: 9.0 Interrupting the script executed withou

(self, line, cell)

    @script_args
    @cell_magic("script")
    def shebang(self, line, cell):
        """Run a cell via a shell command

        The `%%script` line is like the #! line of script,
        specifying a program (bash, perl, ruby, etc.) with which to run.

        The rest of the cell is run by that program.

        .. versionchanged:: 9.0
          Interrupting the script executed without `--bg` will end in
          raising an exception (unless `--no-raise-error` is passed).

        Examples
        --------
        ::

            In [1]: %%script bash
               ...: for i in 1 2 3; do
               ...:   echo $i
               ...: done
            1
            2
            3
        """

        # Create the event loop in which to run script magics
        # this operates on a background thread
        if self.event_loop is None:
            if sys.platform == "win32":
                # don't override the current policy,
                # just create an event loop
                event_loop = asyncio.WindowsProactorEventLoopPolicy().new_event_loop()
            else:
                event_loop = asyncio.new_event_loop()
            self.event_loop = event_loop

            # start the loop in a background thread
            asyncio_thread = Thread(target=event_loop.run_forever, daemon=True)
            asyncio_thread.start()
        else:
            event_loop = self.event_loop

        def in_thread(coro):
            """Call a coroutine on the asyncio thread"""
            return asyncio.run_coroutine_threadsafe(coro, event_loop).result()

        async def _readchunk(stream):
            try:
                return await stream.read(100)
            except asyncio.exceptions.IncompleteReadError as e:
                return e.partial
            except asyncio.exceptions.LimitOverrunError as e:
                return await stream.read(e.consumed)

        async def _handle_stream(stream, stream_arg, file_object):
            should_break = False
            decoder = getincrementaldecoder("utf-8")(errors="replace")
            while True: