vistir.contextmanagers module

vistir.contextmanagers.temp_environ()[source]

Allow the ability to set os.environ temporarily

vistir.contextmanagers.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.contextmanagers.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.contextmanagers.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.contextmanagers.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.contextmanagers.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.contextmanagers.dummy_spinner(spin_type, text, **kwargs)[source]
vistir.contextmanagers.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.contextmanagers.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'