I/O & System Utilities

Input/output operations and system interaction utilities.

iolib

I/O operations: CSV rendering, zipped CSV handling, iterable-to-stream conversion, JSON byteification, print suppression.

render_csv(rows, dialect=<class 'csv.excel'>)

Render rows as CSV string.

Parameters:

• rows – Iterable of rows to render.

• dialect – CSV dialect to use (default: csv.excel).

Returns:

CSV formatted string.

Return type:

str

Example:

>>> render_csv([['a', 'b'], ['1', '2']])
'a,b\r\n1,2\r\n'

class CsvZip

Bases: ZipFile

Zipped CSV file that handles file permissions correctly on DOS.

Example:

>>> cz = CsvZip()
>>> cz.writecsv('test', [['a', 'b'], ['1', '2']])
>>> len(cz.value) > 0
True

Note

See http://stackoverflow.com/q/279945/424380

property value

writecsv(filename, data)

iterable_to_stream(iterable, buffer_size=8192)

Convert an iterable that yields bytestrings to a read-only input stream.

Parameters:

• iterable – Iterable yielding bytestrings.

• buffer_size (int) – Buffer size for the stream.

Returns:

BufferedReader stream.

Example:

>>> stream = iterable_to_stream([b'hello', b' ', b'world'])
>>> stream.read()
b'hello world'

Note

See https://stackoverflow.com/a/20260030

stream(func)

Decorator that converts first streamable input param to a stream.

Parameters:

func – Function to wrap.

Returns:

Wrapped function.

json_load_byteified(file_handle)

Parse ASCII encoded JSON from file handle.

Parameters:

file_handle – File handle to read from.

Returns:

Parsed JSON data.

json_loads_byteified(json_text)

Parse ASCII encoded JSON from text string.

Parameters:

json_text (str) – JSON text to parse.

Returns:

Parsed JSON data.

Example:

>>> json_loads_byteified('{"foo": "bar"}')
{'foo': 'bar'}
>>> json_loads_byteified('{"foo": "bar", "things": [7, {"qux": "baz", "moo": {"cow": ["milk"]}}]}')
{'foo': 'bar', 'things': [7, {'qux': 'baz', 'moo': {'cow': ['milk']}}]}

suppress_print()

Context manager to suppress stdout (print statements).

Useful when third-party code includes unwanted print statements.

Example:

>>> with suppress_print():
...     print("This won't appear")

wrap_suppress_print(func)

Decorator version of suppress_print context manager.

Parameters:

func – Function to wrap.

Returns:

Wrapped function with suppressed stdout.

Example:

>>> @wrap_suppress_print
... def noisy():
...     print("This won't appear")
...     return 42
>>> noisy()
42

stream

Stream utilities: YAML/JSON conversion, binary/text handling, checksum calculation, stream decorators.

is_tty()

Check if running in a terminal.

Returns True if: - Running under pytest (always True for testing) - stdin is a TTY (interactive input) - stderr is a TTY (logger output visible in terminal)

Returns:

True if in a terminal environment.

Return type:

bool

stream_is_tty(somestream)

Check if a stream is running in a terminal.

Parameters:

somestream – Stream to check (typically sys.stdout).

Returns:

True if stream is a TTY.

Return type:

bool

Example:

>>> import io
>>> stream_is_tty(io.StringIO())
False

proc

Process utilities: finding processes by name/port, killing processes, process management.

process_by_name(name)

Find processes by name that are listening on a port.

Parameters:

name (str) – Process name to search for.

Yields:

psutil.Process objects matching the name with LISTEN connections.

See also

See tests/test_proc.py for usage examples.

process_by_name_and_port(name, port)

Find a process by name listening on a specific port.

Parameters:

• name (str) – Process name to search for.

• port (int) – Port number to match.

Returns:

psutil.Process if found, None otherwise.

See also

See tests/test_proc.py for usage examples.

kill_proc(name=None, version=None, dry_run=False, use_terminate=False)

Kill processes matching name and/or version.

Parameters:

• name (str) – Process name pattern (regex).

• version (str) – Version string to match in command line.

• dry_run (bool) – If True, just check for matches without killing.

• use_terminate (bool) – If True, use terminate instead of kill.

Returns:

True if matching processes were found.

Return type:

bool

See also

See tests/test_proc.py for usage examples.

signals

Signal handling: DelayedKeyboardInterrupt context manager for deferring Ctrl+C during critical sections, signal translation map.

class DelayedKeyboardInterrupt(propagate_to_forked_processes=None)

Bases: object

Context manager that suppresses SIGINT & SIGTERM during a block.

Signal handlers are called on exit from the block.

Note

Inspired by https://stackoverflow.com/a/21919644

Parameters:

propagate_to_forked_processes – Controls behavior in forked processes: - True: Same behavior as parent process - False: Use original signal handler - None: Ignore signals (default)

mime

MIME type utilities for guessing file extensions and content types.

guess_type(url)

Guess mimetype from a URL or filename.

Parameters:

url (str) – URL or filename to examine.

Returns:

Guessed mimetype or None.

Return type:

str

Example:

>>> guess_type('document.pdf')
'application/pdf'
>>> guess_type('image.jpg')
'image/jpeg'

guess_extension(mimetype)

Guess file extension for a mimetype.

Parameters:

mimetype (str) – The mimetype to look up.

Returns:

File extension (including dot) or None.

Return type:

str

Example:

>>> guess_extension('image/jpeg')
'.jpg'

magic_mime_from_buffer(buffer)

Detect mimetype using the magic library.

Parameters:

buffer (bytes) – Buffer from header of file.

Returns:

The detected mimetype.

Return type:

str