Miscellaneous utilities for dealing with filesystems, paths, projects, subprocesses, and more.
Project description
🐉 Installation
Install from PyPI:
$ pipenv install vistir
Install from Github:
$ pipenv install -e git+https://github.com/sarugaku/vistir.git#egg=vistir
🐉 Summary
vistir is a library full of utility functions designed to make life easier. Here are some of the places where these functions are used:
🐉 Usage
Importing a utility
You can import utilities directly from vistir:
from vistir import cd
with cd('/path/to/somedir'):
do_stuff_in('somedir')
🐉 Functionality
vistir provides several categories of functionality, including:
Backports
Compatibility Shims
Context Managers
Miscellaneous Utilities
Path Utilities
🐉 Compatibility Shims
Shims are provided for full API compatibility from python 2.7 through 3.7 for the following:
weakref.finalize
functools.partialmethod (via vistir.backports.functools.partialmethod)
tempfile.TemporaryDirectory (via vistir.backports.tempfile.TemporaryDirectory)
tempfile.NamedTemporaryFile (via vistir.backports.tempfile.NamedTemporaryFile)
vistir.compat.Path
vistir.compat.get_terminal_size
vistir.compat.JSONDecodeError
vistir.compat.ResourceWarning
vistir.compat.FileNotFoundError
vistir.compat.PermissionError
vistir.compat.IsADirectoryError
The following additional functions are provided for encoding strings to the filesystem default encoding:
vistir.compat.fs_str
vistir.compat.to_native_string
vistir.compat.fs_encode
vistir.compat.fs_decode
🐉 Context Managers
vistir provides the following context managers as utility contexts:
vistir.contextmanagers.atomic_open_for_write
vistir.contextmanagers.cd
vistir.contextmanagers.open_file
vistir.contextmanagers.replaced_stream
vistir.contextmanagers.replaced_streams
vistir.contextmanagers.spinner
vistir.contextmanagers.temp_environ
vistir.contextmanagers.temp_path
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()
this is some test 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)
replaced_stream
A context manager to temporarily swap out stream_name with a stream wrapper. This will capture the stream output and prevent it from being written as normal.
>>> orig_stdout = sys.stdout
>>> with replaced_stream("stdout") as stdout:
... sys.stdout.write("hello")
... assert stdout.getvalue() == "hello"
... assert orig_stdout.getvalue() != "hello"
>>> sys.stdout.write("hello")
'hello'
replaced_streams
Temporarily replaces both sys.stdout and sys.stderr and captures anything written to these respective targets.
>>> 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'
spinner
A context manager for wrapping some actions with a threaded, interrupt-safe spinner. The spinner is fully compatible with all terminals (you can use bouncingBar on non-utf8 terminals) and will allow you to update the text of the spinner itself by simply setting spinner.text or write lines to the screen above the spinner by using spinner.write(line). Success text can be indicated using spinner.ok("Text") and failure text can be indicated with spinner.fail("Fail text").
>>> lines = ["a", "b"]
>>> with vistir.contextmanagers.spinner(spinner_name="dots", text="Running...", handler_map={}, nospin=False) as sp:
for line in lines:
sp.write(line + "\n")
while some_variable = some_queue.pop():
sp.text = "Consuming item: %s" % some_variable
if success_condition:
sp.ok("Succeeded!")
else:
sp.fail("Failed!")
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']
🐉 Cursor Utilities
The following Cursor utilities are available to manipulate the console cursor:
vistir.cursor.hide_cursor
vistir.cursor.show_cursor
hide_cursor
Hide the console cursor in the given stream.
>>> vistir.cursor.hide_cursor(stream=sys.stdout)
show_cursor
Show the console cursor in the given stream.
>>> vistir.cursor.show_cursor(stream=sys.stdout)
🐉 Miscellaneous Utilities
The following Miscellaneous utilities are available as helper methods:
vistir.misc.shell_escape
vistir.misc.unnest
vistir.misc.dedup
vistir.misc.run
vistir.misc.load_path
vistir.misc.partialclass
vistir.misc.to_text
vistir.misc.to_bytes
vistir.misc.divide
vistir.misc.take
vistir.misc.chunked
vistir.misc.decode_for_output
vistir.misc.get_canonical_encoding_name
vistir.misc.get_wrapped_stream
vistir.misc.StreamWrapper
vistir.misc.get_text_stream
vistir.misc.replace_with_text_stream
vistir.misc.get_text_stdin
vistir.misc.get_text_stdout
vistir.misc.get_text_stderr
vistir.misc.echo
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.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'}
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'
chunked
Splits an iterable up into groups of the specified length, per more itertools. Returns an iterable.
This example will create groups of chunk size 5, which means there will be 6 groups.
>>> chunked_iterable = vistir.misc.chunked(5, range(30))
>>> for chunk in chunked_iterable:
... add_to_some_queue(chunk)
take
Take elements from the supplied iterable without consuming it.
>>> iterable = range(30)
>>> first_10 = take(10, iterable)
>>> [i for i in first_10]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> [i for i in iterable]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
divide
Splits an iterable up into the specified number of groups, per more itertools. Returns an iterable.
>>> iterable = range(30)
>>> groups = []
>>> for grp in vistir.misc.divide(3, iterable):
... groups.append(grp)
>>> groups
[<tuple_iterator object at 0x7fb7966006a0>, <tuple_iterator object at 0x7fb796652780>, <tuple_iterator object at 0x7fb79650a2b0>]
decode_for_output
Converts an arbitrary text input to output which is encoded for printing to terminal outputs using the system preferred locale using locale.getpreferredencoding(False) with some additional hackery on linux systems.
get_canonical_encoding_name
Given an encoding name, get the canonical name from a codec lookup.
>>> vistir.misc.get_canonical_encoding_name("utf8")
"utf-8"
get_wrapped_stream
Given a stream, wrap it in a StreamWrapper instance and return the wrapped stream.
>>> stream = sys.stdout
>>> wrapped_stream = vistir.misc.get_wrapped_stream(sys.stdout)
>>> wrapped_stream.write("unicode\u0141")
>>> wrapped_stream.seek(0)
>>> wrapped_stream.read()
"unicode\u0141"
StreamWrapper
A stream wrapper and compatibility class for handling wrapping file-like stream objects which may be used in place of sys.stdout and other streams.
>>> wrapped_stream = vistir.misc.StreamWrapper(sys.stdout, encoding="utf-8", errors="replace", line_buffering=True)
>>> wrapped_stream = vistir.misc.StreamWrapper(io.StringIO(), encoding="utf-8", errors="replace", line_buffering=True)
get_text_stream
An implementation of the StreamWrapper for the purpose of wrapping sys.stdin or sys.stdout.
On Windows, this returns the appropriate handle to the requested output stream.
>>> text_stream = vistir.misc.get_text_stream("stdout")
>>> sys.stdout = text_stream
>>> sys.stdin = vistir.misc.get_text_stream("stdin")
>>> vistir.misc.echo(u"\0499", fg="green")
ҙ
replace_with_text_stream
Given a text stream name, replaces the text stream with a StreamWrapper instance.
>>> vistir.misc.replace_with_text_stream("stdout")
Once invoked, the standard stream in question is replaced with the required wrapper, turning it into a TextIOWrapper compatible stream (which ensures that unicode characters can be written to it).
get_text_stdin
A helper function for calling get_text_stream(“stdin”).
get_text_stdout
A helper function for calling get_text_stream(“stdout”).
get_text_stderr
A helper function for calling get_text_stream(“stderr”).
echo
Writes colored, stream-compatible output to the desired handle (sys.stdout by default).
>>> vistir.misc.echo("some text", fg="green", bg="black", style="bold", err=True) # write to stderr
some text
>>> vistir.misc.echo("some other text", fg="cyan", bg="white", style="underline") # write to stdout
some other text
🐉 Path Utilities
vistir provides utilities for interacting with filesystem paths:
vistir.path.get_converted_relative_path
vistir.path.normalize_path
vistir.path.is_in_path
vistir.path.handle_remove_readonly
vistir.path.is_file_url
vistir.path.is_readonly_path
vistir.path.is_valid_url
vistir.path.mkdir_p
vistir.path.ensure_mkdir_p
vistir.path.create_tracked_tempdir
vistir.path.create_tracked_tempfile
vistir.path.path_to_url
vistir.path.rmtree
vistir.path.safe_expandvars
vistir.path.set_write_bit
vistir.path.url_to_path
vistir.path.walk_up
normalize_path
Return a case-normalized absolute variable-expanded path.
>>> vistir.path.normalize_path("~/${USER}")
/home/user/user
is_in_path
Determine if the provided full path is in the given parent root.
>>> vistir.path.is_in_path("~/.pyenv/versions/3.7.1/bin/python", "${PYENV_ROOT}/versions")
True
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)
create_tracked_tempfile
Creates a tracked temporary file using vistir.compat.NamedTemporaryFile, but creates a weakref.finalize call which will detach on garbage collection to close and delete the file.
>>> temp_file = vistir.path.create_tracked_tempfile(prefix="requirements", suffix="txt")
>>> temp_file.write("some\nstuff")
>>> exit()
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'
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for vistir-0.4.2-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 588b81ce97fb2b7429ae7b0443a28be8e0c85bb1d0ba7a4296775035c4156760 |
|
MD5 | 22605d209963a18442ccebdeb6624444 |
|
BLAKE2b-256 | a207d664bdd27bace31ee7b86118c768c591acd3444f5adadd16836d69b3e616 |