vistir package

vistir.shell_escape(cmd)[source]

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.

vistir.load_path(python)[source]

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/python37.zip',
 '/home/user/.virtualenvs/requirementslib-5MhGuG3C/lib/python3.7',
 '/home/user/.virtualenvs/requirementslib-5MhGuG3C/lib/python3.7/lib-dynload',
 '/home/user/.pyenv/versions/3.7.0/lib/python3.7',
 '/home/user/.virtualenvs/requirementslib-5MhGuG3C/lib/python3.7/site-packages',
 '/home/user/git/requirementslib/src']
vistir.run(cmd, 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.

Parameters:
  • 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.
Returns:

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

Warning

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="https://pypi.org/simple")
>>> source
<class '__main__.Source'>
>>> source(name="pypi")
>>> source.__dict__
mappingproxy({
    '__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='https://pypi.org/simple'
    )
})
>>> new_source = source(name="pypi")
>>> new_source
<__main__.Source object at 0x7f23af189b38>
>>> new_source.__dict__
{'url': 'https://pypi.org/simple', 'verify_ssl': True, 'name': 'pypi'}
vistir.temp_environ()[source]

Allow the ability to set os.environ temporarily

vistir.temp_path()[source]

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)
[
    '/home/user/.pyenv/versions/3.7.0/bin',
    '/home/user/.pyenv/versions/3.7.0/lib/python37.zip',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7/lib-dynload',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7/site-packages'
]
>>> with temp_path():
        sys.path = path_from_virtualenv
        # Running in the context of the path above
        run(["pip", "install", "stuff"])
>>> print(sys.path)
[
    '/home/user/.pyenv/versions/3.7.0/bin',
    '/home/user/.pyenv/versions/3.7.0/lib/python37.zip',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7/lib-dynload',
    '/home/user/.pyenv/versions/3.7.0/lib/python3.7/site-packages'
]
vistir.cd(path)[source]

Context manager to temporarily change working directories

Parameters:path (str) – The directory to move into
>>> print(os.path.abspath(os.curdir))
'/home/user/code/myrepo'
>>> 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))
'/home/user/code/myrepo'
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.

Parameters:
  • 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:
            print(fh.read().strip())

>>> 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.

Parameters:
  • 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
Raises:

ValueError – If link points to a local directory.

Returns:

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.

Parameters:
  • 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()

Note

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.

Parameters:
  • 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)
Returns:

A spinner object which can be manipulated while alive

Return type:

VistirSpinner

Raises:
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.

Parameters:
  • 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.
Returns:

A re-encoded string using the preferred encoding

Return type:

str

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

Force a value to a text-type.

Parameters:
  • 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
Returns:

The unicode representation of the string

Return type:

str

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

Force a value to bytes.

Parameters:
  • 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
Returns:

Corresponding byte representation (for use in filesystem operations)

Return type:

bytes

vistir.take(n, iterable)[source]

Take n elements from the supplied iterable without consuming it.

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

from https://github.com/erikrose/more-itertools/blob/master/more_itertools/recipes.py

vistir.chunked(n, iterable)[source]

Split an iterable into lists of length n.

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

from https://github.com/erikrose/more-itertools/blob/master/more_itertools/more.py

vistir.divide(n, iterable)[source]

split an iterable into n groups, per https://more- itertools.readthedocs.io/en/latest/api.html#grouping.

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

a list of new iterables derived from the original iterable

Return type:

list

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

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

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

A new, wrapped stream

Return type:

StreamWrapper

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.

isatty()[source]

Return whether this is an ‘interactive’ stream.

Return False if it can’t be determined.

write(x)[source]

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

writelines(lines)[source]

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.

vistir.replaced_stream(stream_name)[source]

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")
'hello'
vistir.replaced_streams()[source]

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()
'hello'
>>> stderr.getvalue()
'test'
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
Returns:None
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
Returns:None
Return type:None