Merge pull request #12932 from minrk/pipes-quote

remove use of deprecated pipes module

IPython/core/magics/script.py

@@ -1,321 +1,321 b''
 """Magic functions for running cells in various scripts."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import errno
 import os
 import sys
 import signal
 import time
 import asyncio
 import atexit
 
 from subprocess import CalledProcessError
 
 from IPython.core import magic_arguments
 from IPython.core.magic import  (
     Magics, magics_class, line_magic, cell_magic
 )
 from IPython.lib.backgroundjobs import BackgroundJobManager
 from IPython.utils.process import arg_split
 from traitlets import List, Dict, default
 
 
 #-----------------------------------------------------------------------------
 # Magic implementation classes
 #-----------------------------------------------------------------------------
 
 def script_args(f):
     """single decorator for adding script args"""
     args = [
         magic_arguments.argument(
             '--out', type=str,
             help="""The variable in which to store stdout from the script.
             If the script is backgrounded, this will be the stdout *pipe*,
             instead of the stderr text itself and will not be auto closed.
             """
         ),
         magic_arguments.argument(
             '--err', type=str,
             help="""The variable in which to store stderr from the script.
             If the script is backgrounded, this will be the stderr *pipe*,
             instead of the stderr text itself and will not be autoclosed.
             """
         ),
         magic_arguments.argument(
             '--bg', action="store_true",
             help="""Whether to run the script in the background.
             If given, the only way to see the output of the command is
             with --out/err.
             """
         ),
         magic_arguments.argument(
             '--proc', type=str,
             help="""The variable in which to store Popen instance.
             This is used only when --bg option is given.
             """
         ),
         magic_arguments.argument(
             '--no-raise-error', action="store_false", dest='raise_error',
             help="""Whether you should raise an error message in addition to 
             a stream on stderr if you get a nonzero exit code.
             """
         )
     ]
     for arg in args:
         f = arg(f)
     return f
 
 @magics_class
 class ScriptMagics(Magics):
     """Magics for talking to scripts
     
     This defines a base `%%script` cell magic for running a cell
     with a program in a subprocess, and registers a few top-level
     magics that call %%script with common interpreters.
     """
     script_magics = List(
         help="""Extra script cell magics to define
         
         This generates simple wrappers of `%%script foo` as `%%foo`.
         
         If you want to add script magics that aren't on your path,
         specify them in script_paths
         """,
     ).tag(config=True)
     @default('script_magics')
     def _script_magics_default(self):
         """default to a common list of programs"""
         
         defaults = [
             'sh',
             'bash',
             'perl',
             'ruby',
             'python',
             'python2',
             'python3',
             'pypy',
         ]
         if os.name == 'nt':
             defaults.extend([
                 'cmd',
             ])
         
         return defaults
     
     script_paths = Dict(
         help="""Dict mapping short 'ruby' names to full paths, such as '/opt/secret/bin/ruby'
         
         Only necessary for items in script_magics where the default path will not
         find the right interpreter.
         """
     ).tag(config=True)
     
     def __init__(self, shell=None):
         super(ScriptMagics, self).__init__(shell=shell)
         self._generate_script_magics()
         self.job_manager = BackgroundJobManager()
         self.bg_processes = []
         atexit.register(self.kill_bg_processes)
 
     def __del__(self):
         self.kill_bg_processes()
     
     def _generate_script_magics(self):
         cell_magics = self.magics['cell']
         for name in self.script_magics:
             cell_magics[name] = self._make_script_magic(name)
     
     def _make_script_magic(self, name):
         """make a named magic, that calls %%script with a particular program"""
         # expand to explicit path if necessary:
         script = self.script_paths.get(name, name)
         
         @magic_arguments.magic_arguments()
         @script_args
         def named_script_magic(line, cell):
             # if line, add it as cl-flags
             if line:
                 line = "%s %s" % (script, line)
             else:
                 line = script
             return self.shebang(line, cell)
         
         # write a basic docstring:
         named_script_magic.__doc__ = \
         """%%{name} script magic
         
         Run cells with {script} in a subprocess.
         
         This is a shortcut for `%%script {script}`
         """.format(**locals())
         
         return named_script_magic
     
     @magic_arguments.magic_arguments()
     @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.
         
         Examples
         --------
         ::
         
             In [1]: %%script bash
                ...: for i in 1 2 3; do
                ...:   echo $i
                ...: done
             1
             2
             3
         """
 
         async def _handle_stream(stream, stream_arg, file_object):
             while True:
                 line = (await stream.readline()).decode("utf8")
                 if not line:
                     break
                 if stream_arg:
                     self.shell.user_ns[stream_arg] = line
                 else:
                     file_object.write(line)
                     file_object.flush()
 
         async def _stream_communicate(process, cell):
             process.stdin.write(cell)
             process.stdin.close()
             stdout_task = asyncio.create_task(
                 _handle_stream(process.stdout, args.out, sys.stdout)
             )
             stderr_task = asyncio.create_task(
                 _handle_stream(process.stderr, args.err, sys.stderr)
             )
             await asyncio.wait([stdout_task, stderr_task])
             await process.wait()
 
         if sys.platform.startswith("win"):
             asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
         loop = asyncio.get_event_loop()
 
         argv = arg_split(line, posix=not sys.platform.startswith("win"))
         args, cmd = self.shebang.parser.parse_known_args(argv)
         try:
             p = loop.run_until_complete(
                 asyncio.create_subprocess_exec(
                     *cmd,
                     stdout=asyncio.subprocess.PIPE,
                     stderr=asyncio.subprocess.PIPE,
-                    stdin=asyncio.subprocess.PIPE
+                    stdin=asyncio.subprocess.PIPE,
                 )
             )
         except OSError as e:
             if e.errno == errno.ENOENT:
                 print("Couldn't find program: %r" % cmd[0])
                 return
             else:
                 raise
         
         if not cell.endswith('\n'):
             cell += '\n'
         cell = cell.encode('utf8', 'replace')
         if args.bg:
             self.bg_processes.append(p)
             self._gc_bg_processes()
             to_close = []
             if args.out:
                 self.shell.user_ns[args.out] = p.stdout
             else:
                 to_close.append(p.stdout)
             if args.err:
                 self.shell.user_ns[args.err] = p.stderr
             else:
                 to_close.append(p.stderr)
             self.job_manager.new(self._run_script, p, cell, to_close, daemon=True)
             if args.proc:
                 self.shell.user_ns[args.proc] = p
             return
         
         try:
             loop.run_until_complete(_stream_communicate(p, cell))
         except KeyboardInterrupt:
             try:
                 p.send_signal(signal.SIGINT)
                 time.sleep(0.1)
                 if p.returncode is not None:
                     print("Process is interrupted.")
                     return
                 p.terminate()
                 time.sleep(0.1)
                 if p.returncode is not None:
                     print("Process is terminated.")
                     return
                 p.kill()
                 print("Process is killed.")
             except OSError:
                 pass
             except Exception as e:
                 print("Error while terminating subprocess (pid=%i): %s" % (p.pid, e))
             return
         if args.raise_error and p.returncode!=0:
             # If we get here and p.returncode is still None, we must have
             # killed it but not yet seen its return code. We don't wait for it,
             # in case it's stuck in uninterruptible sleep. -9 = SIGKILL
             rc = p.returncode or -9
             raise CalledProcessError(rc, cell)
     
     def _run_script(self, p, cell, to_close):
         """callback for running the script in the background"""
         p.stdin.write(cell)
         p.stdin.close()
         for s in to_close:
             s.close()
         p.wait()
 
     @line_magic("killbgscripts")
     def killbgscripts(self, _nouse_=''):
         """Kill all BG processes started by %%script and its family."""
         self.kill_bg_processes()
         print("All background processes were killed.")
 
     def kill_bg_processes(self):
         """Kill all BG processes which are still running."""
         if not self.bg_processes:
             return
         for p in self.bg_processes:
             if p.returncode is None:
                 try:
                     p.send_signal(signal.SIGINT)
                 except:
                     pass
         time.sleep(0.1)
         self._gc_bg_processes()
         if not self.bg_processes:
             return
         for p in self.bg_processes:
             if p.returncode is None:
                 try:
                     p.terminate()
                 except:
                     pass
         time.sleep(0.1)
         self._gc_bg_processes()
         if not self.bg_processes:
             return
         for p in self.bg_processes:
             if p.returncode is None:
                 try:
                     p.kill()
                 except:
                     pass
         self._gc_bg_processes()
 
     def _gc_bg_processes(self):
         self.bg_processes = [p for p in self.bg_processes if p.returncode is None]

IPython/lib/editorhooks.py

@@ -1,128 +1,127 b''
 """ 'editor' hooks for common editors that work well with ipython
 
 They should honor the line number argument, at least.
 
 Contributions are *very* welcome.
 """
 
 import os
-import pipes
 import shlex
 import subprocess
 import sys
 
 from IPython import get_ipython
 from IPython.core.error import TryNext
 from IPython.utils import py3compat
 
 
 def install_editor(template, wait=False):
     """Installs the editor that is called by IPython for the %edit magic.
 
     This overrides the default editor, which is generally set by your EDITOR
     environment variable or is notepad (windows) or vi (linux). By supplying a
     template string `run_template`, you can control how the editor is invoked
     by IPython -- (e.g. the format in which it accepts command line options)
 
     Parameters
     ----------
     template : basestring
         run_template acts as a template for how your editor is invoked by
         the shell. It should contain '{filename}', which will be replaced on
         invocation with the file name, and '{line}', $line by line number
         (or 0) to invoke the file with.
     wait : bool
         If `wait` is true, wait until the user presses enter before returning,
         to facilitate non-blocking editors that exit immediately after
         the call.
     """
 
     # not all editors support $line, so we'll leave out this check
     # for substitution in ['$file', '$line']:
     #    if not substitution in run_template:
     #        raise ValueError(('run_template should contain %s'
     #        ' for string substitution. You supplied "%s"' % (substitution,
     #            run_template)))
 
     def call_editor(self, filename, line=0):
         if line is None:
             line = 0
-        cmd = template.format(filename=pipes.quote(filename), line=line)
+        cmd = template.format(filename=shlex.quote(filename), line=line)
         print(">", cmd)
-        # pipes.quote doesn't work right on Windows, but it does after splitting
+        # shlex.quote doesn't work right on Windows, but it does after splitting
         if sys.platform.startswith('win'):
             cmd = shlex.split(cmd)
         proc = subprocess.Popen(cmd, shell=True)
         if proc.wait() != 0:
             raise TryNext()
         if wait:
             py3compat.input("Press Enter when done editing:")
 
     get_ipython().set_hook('editor', call_editor)
     get_ipython().editor = template
 
 
 # in these, exe is always the path/name of the executable. Useful
 # if you don't have the editor directory in your path
 def komodo(exe=u'komodo'):
     """ Activestate Komodo [Edit] """
     install_editor(exe + u' -l {line} {filename}', wait=True)
 
 
 def scite(exe=u"scite"):
     """ SciTE or Sc1 """
     install_editor(exe + u' {filename} -goto:{line}')
 
 
 def notepadplusplus(exe=u'notepad++'):
     """ Notepad++ http://notepad-plus.sourceforge.net """
     install_editor(exe + u' -n{line} {filename}')
 
 
 def jed(exe=u'jed'):
     """ JED, the lightweight emacsish editor """
     install_editor(exe + u' +{line} {filename}')
 
 
 def idle(exe=u'idle'):
     """ Idle, the editor bundled with python
 
     Parameters
     ----------
     exe : str, None
         If none, should be pretty smart about finding the executable.
     """
     if exe is None:
         import idlelib
         p = os.path.dirname(idlelib.__filename__)
         # i'm not sure if this actually works. Is this idle.py script
         # guaranteed to be executable?
         exe = os.path.join(p, 'idle.py')
     install_editor(exe + u' {filename}')
 
 
 def mate(exe=u'mate'):
     """ TextMate, the missing editor"""
     # wait=True is not required since we're using the -w flag to mate
     install_editor(exe + u' -w -l {line} {filename}')
 
 
 # ##########################################
 # these are untested, report any problems
 # ##########################################
 
 
 def emacs(exe=u'emacs'):
     install_editor(exe + u' +{line} {filename}')
 
 
 def gnuclient(exe=u'gnuclient'):
     install_editor(exe + u' -nw +{line} {filename}')
 
 
 def crimson_editor(exe=u'cedt.exe'):
     install_editor(exe + u' /L:{line} {filename}')
 
 
 def kate(exe=u'kate'):
     install_editor(exe + u' -u -l {line} {filename}')