vistir package


Escape strings for use in Popen() and run().

This is a passthrough method for instantiating a Script object which can be used to escape commands to output as a single string.


Load the sys.path from the given python executable’s environment as json.

Parameters:python (str) – Path to a valid python executable
Returns:A python representation of the sys.path value of the given python executable.
Return type:list
>>> load_path("/home/user/.virtualenvs/requirementslib-5MhGuG3C/bin/python")
['', '/home/user/.virtualenvs/requirementslib-5MhGuG3C/lib/',
 '/home/user/git/requirementslib/src'], env=None, return_object=False, block=True, cwd=None, verbose=False, nospin=False, spinner_name=None, combine_stderr=True, display_limit=200, write_to_stdout=True, encoding='utf-8')[source]

Use subprocess.Popen to get the output of a command and decode it.

  • cmd (list) – A list representing the command you want to run.
  • env (dict) – Additional environment settings to pass through to the subprocess.
  • return_object (bool) – When True, returns the whole subprocess instance
  • block (bool) – When False, returns a potentially still-running subprocess.Popen instance
  • cwd (str) – Current working directory context to use for spawning the subprocess.
  • verbose (bool) – Whether to print stdout in real time when non-blocking.
  • nospin (bool) – Whether to disable the cli spinner.
  • spinner_name (str) – The name of the spinner to use if enabled, defaults to bouncingBar
  • combine_stderr (bool) – Optionally merge stdout and stderr in the subprocess, false if nonblocking.
  • dispay_limit (int) – The max width of output lines to display when using a spinner.
  • write_to_stdout (bool) – Whether to write to stdout when using a spinner, defaults to True.

A 2-tuple of (output, error) or a subprocess.Popen object.


Merging standard out and standard error in a nonblocking subprocess can cause errors in some cases and may not be ideal. Consider disabling this functionality.

vistir.partialclass(cls, *args, **kwargs)[source]

Returns a partially instantiated class.

Returns:A partial class instance
Return type:cls
>>> source = partialclass(Source, url="")
>>> source
<class '__main__.Source'>
>>> source(name="pypi")
>>> source.__dict__
    '__module__': '__main__',
    '__dict__': <attribute '__dict__' of 'Source' objects>,
    '__weakref__': <attribute '__weakref__' of 'Source' objects>,
    '__doc__': None,
    '__init__': functools.partialmethod(
        <function Source.__init__ at 0x7f23af429bf8>, , url=''
>>> new_source = source(name="pypi")
>>> new_source
<__main__.Source object at 0x7f23af189b38>
>>> new_source.__dict__
{'url': '', 'verify_ssl': True, 'name': 'pypi'}

Allow the ability to set os.environ temporarily


A context manager which allows the ability to set sys.path temporarily

>>> path_from_virtualenv = load_path("/path/to/venv/bin/python")
>>> print(sys.path)
>>> with temp_path():
        sys.path = path_from_virtualenv
        # Running in the context of the path above
        run(["pip", "install", "stuff"])
>>> print(sys.path)

Context manager to temporarily change working directories

Parameters:path (str) – The directory to move into
>>> print(os.path.abspath(os.curdir))
>>> with cd("/home/user/code/otherdir/subdir"):
...     print("Changed directory: %s" % os.path.abspath(os.curdir))
Changed directory: /home/user/code/otherdir/subdir
>>> print(os.path.abspath(os.curdir))
vistir.atomic_open_for_write(target, binary=False, newline=None, encoding=None)[source]

Atomically open target for writing.

This is based on Lektor’s atomic_open() utility, but simplified a lot to handle only writing, and skip many multi-process/thread edge cases handled by Werkzeug.

  • target (str) – Target filename to write
  • binary (bool) – Whether to open in binary mode, default False
  • newline (Optional[str]) – The newline character to use when writing, determined from system if not supplied.
  • encoding (Optional[str]) – The encoding to use when writing, defaults to system encoding.

How this works:

  • Create a temp file (in the same directory of the actual target), and yield for surrounding code to write to it.
  • If some thing goes wrong, try to remove the temp file. The actual target is not touched whatsoever.
  • If everything goes well, close the temp file, and replace the actual target with this new file.
>>> fn = "test_file.txt"
>>> def read_test_file(filename=fn):
        with open(filename, 'r') as fh:

>>> with open(fn, "w") as fh:
        fh.write("this is some test text")
>>> read_test_file()
this is some test text

>>> def raise_exception_while_writing(filename):
        with open(filename, "w") as fh:
            fh.write("writing some new text")
            raise RuntimeError("Uh oh, hope your file didn't get overwritten")

>>> raise_exception_while_writing(fn)
Traceback (most recent call last):
RuntimeError: Uh oh, hope your file didn't get overwritten
>>> read_test_file()
writing some new text

# Now try with vistir
>>> def raise_exception_while_writing(filename):
        with vistir.contextmanagers.atomic_open_for_write(filename) as fh:
            fh.write("Overwriting all the text from before with even newer text")
            raise RuntimeError("But did it get overwritten now?")

>>> raise_exception_while_writing(fn)
    Traceback (most recent call last):
    RuntimeError: But did it get overwritten now?

>>> read_test_file()
    writing some new text
vistir.open_file(link, session=None, stream=True)[source]

Open local or remote file for reading.

  • link (pip._internal.index.Link) – A link object from resolving dependencies with pip, or else a URL.
  • session (Optional[Session]) – A Session instance
  • stream (bool) – Whether to stream the content if remote, default True

ValueError – If link points to a local directory.


a context manager to the opened file-like object

vistir.rmtree(directory: str, ignore_errors: bool = False, onerror: Optional[Callable] = None) → None[source]

Stand-in for rmtree() with additional error-handling.

This version of rmtree handles read-only paths, especially in the case of index files written by certain source control systems.

  • directory (str) – The target directory to remove
  • ignore_errors (bool) – Whether to ignore errors, defaults to False
  • onerror (func) – An error handling function, defaults to handle_remove_readonly()


Setting ignore_errors=True may cause this to silently fail to delete the path

vistir.mkdir_p(newdir, mode=511)[source]
vistir.spinner(spinner_name=None, start_text=None, handler_map=None, nospin=False, write_to_stdout=True)[source]

Get a spinner object or a dummy spinner to wrap a context.

  • spinner_name (str) – A spinner type e.g. “dots” or “bouncingBar” (default: {“bouncingBar”})
  • start_text (str) – Text to start off the spinner with (default: {None})
  • handler_map (dict) – Handler map for signals to be handled gracefully (default: {None})
  • nospin (bool) – If true, use the dummy spinner (default: {False})
  • write_to_stdout (bool) – Writes to stdout if true, otherwise writes to stderr (default: True)

A spinner object which can be manipulated while alive

Return type:


RuntimeError – Raised if the spinner extra is not installed
vistir.create_spinner(*args, **kwargs)[source]
vistir.create_tracked_tempdir(*args, **kwargs)[source]

Create a tracked temporary directory.

This uses TemporaryDirectory, but does not remove the directory when the return value goes out of scope, instead registers a handler to cleanup on program exit.

The return value is the path to the created directory.

vistir.create_tracked_tempfile(*args, **kwargs)[source]

Create a tracked temporary file.

This uses the NamedTemporaryFile construct, but does not remove the file until the interpreter exits.

The return value is the file object.

vistir.decode_for_output(output, target_stream=None, translation_map=None)[source]

Given a string, decode it for output to a terminal.

  • output (str) – A string to print to a terminal
  • target_stream – A stream to write to, we will encode to target this stream if possible.
  • translation_map (dict) – A mapping of unicode character ordinals to replacement strings.

A re-encoded string using the preferred encoding

Return type:


vistir.to_text(string, encoding='utf-8', errors=None)[source]

Force a value to a text-type.

  • string (str or bytes unicode) – Some input that can be converted to a unicode representation.
  • encoding – The encoding to use for conversions, defaults to “utf-8”
  • encoding – str, optional

The unicode representation of the string

Return type:


vistir.to_bytes(string, encoding='utf-8', errors=None)[source]

Force a value to bytes.

  • string (str or bytes unicode or a memoryview subclass) – Some input that can be converted to a bytes.
  • encoding – The encoding to use for conversions, defaults to “utf-8”
  • encoding – str, optional

Corresponding byte representation (for use in filesystem operations)

Return type:


vistir.take(n, iterable)[source]

Take n elements from the supplied iterable without consuming it.

  • n (int) – Number of unique groups
  • iterable (iter) – An iterable to split up


vistir.chunked(n, iterable)[source]

Split an iterable into lists of length n.

  • n (int) – Number of unique groups
  • iterable (iter) – An iterable to split up


vistir.divide(n, iterable)[source]

split an iterable into n groups, per https://more-

  • n (int) – Number of unique groups
  • iterable (iter) – An iterable to split up

a list of new iterables derived from the original iterable

Return type:


vistir.get_wrapped_stream(stream, encoding=None, errors='replace')[source]

Given a stream, wrap it in a StreamWrapper instance and return the wrapped stream.

  • stream – A stream instance to wrap
  • encoding (str) – The encoding to use for the stream
  • errors (str) – The error handler to use, default “replace”

A new, wrapped stream

Return type:


class vistir.StreamWrapper(stream, encoding, errors, line_buffering=True, **kwargs)[source]

Bases: _io.TextIOWrapper

This wrapper class will wrap a provided stream and supply an interface for compatibility.


Return whether this is an ‘interactive’ stream.

Return False if it can’t be determined.


Write string to stream. Returns the number of characters written (which is always equal to the length of the string).


Write a list of lines to stream.

Line separators are not added, so it is usual for each of the lines provided to have a line separator at the end.


Context manager to temporarily swap out stream_name with a stream wrapper.

Parameters:stream_name (str) – The name of a sys stream to wrap
Returns:A StreamWrapper replacement, temporarily
>>> orig_stdout = sys.stdout
>>> with replaced_stream("stdout") as stdout:
...     sys.stdout.write("hello")
...     assert stdout.getvalue() == "hello"
>>> sys.stdout.write("hello")

Context manager to replace both sys.stdout and sys.stderr using replaced_stream

returns: (stdout, stderr)

>>> import sys
>>> with vistir.contextmanagers.replaced_streams() as streams:
>>>     stdout, stderr = streams
>>>     sys.stderr.write("test")
>>>     sys.stdout.write("hello")
>>>     assert stdout.getvalue() == "hello"
>>>     assert stderr.getvalue() == "test"
>>> stdout.getvalue()
>>> stderr.getvalue()
vistir.show_cursor(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>)[source]

Show the console cursor on the given stream

Parameters:stream – The name of the stream to get the handle for
Return type:None
vistir.hide_cursor(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>)[source]

Hide the console cursor on the given stream

Parameters:stream – The name of the stream to get the handle for
Return type:None