vistir 0.1.7

atomic_open_for_write

This context manager ensures that a file only gets overwritten if the contents can be successfully written in its place. If you open a file for writing and then fail in the middle under normal circumstances, your original file is already gone.

>>> fn = "test_file.txt"
>>> 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 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

cd

A context manager for temporarily changing the working directory.

>>> os.path.abspath(os.curdir)
'/tmp/test'
>>> with vistir.contextmanagers.cd('/tmp/vistir_test'):
        print(os.path.abspath(os.curdir))
/tmp/vistir_test

open_file

A context manager for streaming file contents, either local or remote. It is recommended to pair this with an iterator which employs a sensible chunk size.

>>> filecontents = b""
    with vistir.contextmanagers.open_file("https://norvig.com/big.txt") as fp:
        for chunk in iter(lambda: fp.read(16384), b""):
            filecontents.append(chunk)
>>> import io
>>> import shutil
>>> filecontents = io.BytesIO(b"")
>>> with vistir.contextmanagers.open_file("https://norvig.com/big.txt") as fp:
        shutil.copyfileobj(fp, filecontents)

temp_environ

Sets a temporary environment context to freely manipulate os.environ which will be reset upon exiting the context.

>>> os.environ['MY_KEY'] = "test"
>>> os.environ['MY_KEY']
'test'
>>> with vistir.contextmanagers.temp_environ():
        os.environ['MY_KEY'] = "another thing"
        print("New key: %s" % os.environ['MY_KEY'])
New key: another thing
>>> os.environ['MY_KEY']
'test'

temp_path

Sets a temporary environment context to freely manipulate sys.path which will be reset upon exiting the context.

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

shell_escape

Escapes a string for use as shell input when passing shell=True to os.Popen.

>>> vistir.misc.shell_escape("/tmp/test/test script.py hello")
'/tmp/test/test script.py hello'

unnest

Unnests nested iterables into a flattened one.

>>> nested_iterable = (1234, (3456, 4398345, (234234)), (2396, (23895750, 9283798, 29384, (289375983275, 293759, 2347, (2098, 7987, 27599)))))
>>> list(vistir.misc.unnest(nested_iterable))
[1234, 3456, 4398345, 234234, 2396, 23895750, 9283798, 29384, 289375983275, 293759, 2347, 2098, 7987, 27599]

dedup

Deduplicates an iterable (like a set, but preserving order).

>>> iterable = ["repeatedval", "uniqueval", "repeatedval", "anotherval", "somethingelse"]
>>> list(vistir.misc.dedup(iterable))
['repeatedval', 'uniqueval', 'anotherval', 'somethingelse']

run

Runs the given command using subprocess.Popen and passing sane defaults.

>>> out, err = vistir.run(["cat", "/proc/version"])
>>> out
'Linux version 4.15.0-27-generic (buildd@lgw01-amd64-044) (gcc version 7.3.0 (Ubuntu 7.3.0-16ubuntu3)) #29-Ubuntu SMP Wed Jul 11 08:21:57 UTC 2018'

load_path

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

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

partialclass

Create a partially instantiated class.

>>> source = partialclass(Source, url="https://pypi-hypernode.com/simple")
>>> new_source = source(name="pypi")
>>> new_source
<__main__.Source object at 0x7f23af189b38>
>>> new_source.__dict__
{'url': 'https://pypi-hypernode.com/simple', 'verify_ssl': True, 'name': 'pypi'}

to_text

Convert arbitrary text-formattable input to text while handling errors.

>>> vistir.misc.to_text(b"these are bytes")
'these are bytes'

to_bytes

Converts arbitrary byte-convertable input to bytes while handling errors.

>>> vistir.misc.to_bytes("this is some text")
b'this is some text'
>>> vistir.misc.to_bytes(u"this is some text")
b'this is some text'

get_converted_relative_path

Convert the supplied path to a relative path (relative to os.curdir)

>>> os.chdir('/home/user/code/myrepo/myfolder')
>>> vistir.path.get_converted_relative_path('/home/user/code/file.zip')
'./../../file.zip'
>>> vistir.path.get_converted_relative_path('/home/user/code/myrepo/myfolder/mysubfolder')
'./mysubfolder'
>>> vistir.path.get_converted_relative_path('/home/user/code/myrepo/myfolder')
'.'

handle_remove_readonly

Error handler for shutil.rmtree.

Windows source repo folders are read-only by default, so this error handler attempts to set them as writeable and then proceed with deletion.

This function will call check vistir.path.is_readonly_path before attempting to call vistir.path.set_write_bit on the target path and try again.

is_file_url

Checks whether the given url is a properly formatted file:// uri.

>>> vistir.path.is_file_url('file:///home/user/somefile.zip')
True
>>> vistir.path.is_file_url('/home/user/somefile.zip')
False

is_readonly_path

Check if a provided path exists and is readonly by checking for bool(path.stat & stat.S_IREAD) and not os.access(path, os.W_OK)

>>> vistir.path.is_readonly_path('/etc/passwd')
True
>>> vistir.path.is_readonly_path('/home/user/.bashrc')
False

is_valid_url

Checks whether a URL is valid and parseable by checking for the presence of a scheme and a netloc.

>>> vistir.path.is_valid_url("https://google.com")
True
>>> vistir.path.is_valid_url("/home/user/somefile")
False

mkdir_p

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

>>> os.mkdir('/tmp/test_dir')
>>> os.listdir('/tmp/test_dir')
[]
>>> vistir.path.mkdir_p('/tmp/test_dir/child/subchild/subsubchild')
>>> os.listdir('/tmp/test_dir/child/subchild')
['subsubchild']

ensure_mkdir_p

A decorator which ensures that the caller function’s return value is created as a directory on the filesystem.

>>> @ensure_mkdir_p
def return_fake_value(path):
    return path
>>> return_fake_value('/tmp/test_dir')
>>> os.listdir('/tmp/test_dir')
[]
>>> return_fake_value('/tmp/test_dir/child/subchild/subsubchild')
>>> os.listdir('/tmp/test_dir/child/subchild')
['subsubchild']

create_tracked_tempdir

Creates a tracked temporary directory using vistir.path.TemporaryDirectory, but does not remove the directory when the return value goes out of scope, instead registers a handler to cleanup on program exit.

>>> temp_dir = vistir.path.create_tracked_tempdir(prefix="test_dir")
>>> assert temp_dir.startswith("test_dir")
True
>>> with vistir.path.create_tracked_tempdir(prefix="test_dir") as temp_dir:
    with io.open(os.path.join(temp_dir, "test_file.txt"), "w") as fh:
        fh.write("this is a test")
>>> os.listdir(temp_dir)

path_to_url

Convert the supplied local path to a file uri.

>>> path_to_url("/home/user/code/myrepo/myfile.zip")
'file:///home/user/code/myrepo/myfile.zip'

rmtree

Stand-in for shutil.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.

>>> vistir.path.rmtree('/tmp/test_dir')
>>> [d for d in os.listdir('/tmp') if 'test_dir' in d]
[]

safe_expandvars

Call os.path.expandvars if value is a string, otherwise do nothing.

>>> os.environ['TEST_VAR'] = "MY_TEST_VALUE"
>>> vistir.path.safe_expandvars("https://myuser:${TEST_VAR}@myfakewebsite.com")
'https://myuser:MY_TEST_VALUE@myfakewebsite.com'

set_write_bit

Set read-write permissions for the current user on the target path. Fail silently if the path doesn’t exist.

>>> vistir.path.set_write_bit('/path/to/some/file')
>>> with open('/path/to/some/file', 'w') as fh:
        fh.write("test text!")

url_to_path

Convert a valid file url to a local filesystem path. Follows logic taken from pip.

>>> vistir.path.url_to_path("file:///home/user/somefile.zip")
'/home/user/somefile.zip'