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)[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 contect 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 standarad 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, ignore_errors=False, onerror=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]

Recursively creates the target directory and all of its parents if they do not already exist. Fails silently if they do.

Parameters:newdir (str) – The directory path to ensure
Raises:OSError if a file is encountered along the way
class vistir.TemporaryDirectory(suffix='', prefix=None, dir=None)[source]

Bases: object

Create and return a temporary directory. This has the same behavior as mkdtemp but can be used as a context manager. For example:

with TemporaryDirectory() as tmpdir:

Upon exiting the context, the directory and everything contained in it are removed.

cleanup()[source]
vistir.NamedTemporaryFile(mode='w+b', buffering=-1, encoding=None, newline=None, suffix=None, prefix=None, dir=None, delete=True)[source]

Create and return a temporary file. Arguments: ‘prefix’, ‘suffix’, ‘dir’ – as for mkstemp. ‘mode’ – the mode argument to io.open (default “w+b”). ‘buffering’ – the buffer size argument to io.open (default -1). ‘encoding’ – the encoding argument to io.open (default None) ‘newline’ – the newline argument to io.open (default None) ‘delete’ – whether the file is deleted on close (default True). The file is created as mkstemp() would do it.

Returns an object with a file-like interface; the name of the file is accessible as its ‘name’ attribute. The file will be automatically deleted when it is closed unless the ‘delete’ argument is set to False.

class vistir.partialmethod(func, *args, **keywords)[source]

Bases: object

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

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.to_native_string(string)[source]
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

class vistir.StringIO

Bases: _io._TextIOBase

Text I/O implementation using an in-memory buffer.

The initial_value argument sets the value of object. The newline argument is like the one of TextIOWrapper’s constructor.

close()

Close the IO object.

Attempting any further operation after the object is closed will raise a ValueError.

This method has no effect if the file is already closed.

closed
getvalue()

Retrieve the entire contents of the object.

line_buffering
newlines

Line endings translated so far.

Only line endings translated during reading are considered.

Subclasses should override.

read()

Read at most size characters, returned as a string.

If the argument is negative or omitted, read until EOF is reached. Return an empty string at EOF.

readable()

Returns True if the IO object can be read.

readline()

Read until newline or EOF.

Returns an empty string if EOF is hit immediately.

seek()

Change stream position.

Seek to character offset pos relative to position indicated by whence:
0 Start of stream (the default). pos should be >= 0; 1 Current position - pos must be 0; 2 End of stream - pos must be 0.

Returns the new absolute position.

seekable()

Returns True if the IO object can be seeked.

tell()

Tell the current file position.

truncate()

Truncate size to pos.

The pos argument defaults to the current file position, as returned by tell(). The current file position is unchanged. Returns the new absolute position.

writable()

Returns True if the IO object can be written.

write()

Write string to file.

Returns the number of characters written, which is always equal to the length of the string.

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