qemu.aqmp.util module

Miscellaneous Utilities

This module provides asyncio utilities and compatibility wrappers for Python 3.6 to provide some features that otherwise become available in Python 3.7+.

Various logging and debugging utilities are also provided, such as exception_summary() and pretty_traceback(), used primarily for adding information into the logging stream.

async qemu.aqmp.util.flush(writer: asyncio.streams.StreamWriter) → None

Utility function to ensure a StreamWriter is fully drained.

asyncio.StreamWriter.drain only promises we will return to below the “high-water mark”. This function ensures we flush the entire buffer – by setting the high water mark to 0 and then calling drain. The flow control limits are restored after the call is completed.

qemu.aqmp.util.upper_half(func: qemu.aqmp.util.T) → qemu.aqmp.util.T

Do-nothing decorator that annotates a method as an “upper-half” method.

These methods must not call bottom-half functions directly, but can schedule them to run.

qemu.aqmp.util.bottom_half(func: qemu.aqmp.util.T) → qemu.aqmp.util.T

Do-nothing decorator that annotates a method as a “bottom-half” method.

These methods must take great care to handle their own exceptions whenever possible. If they go unhandled, they will cause termination of the loop.

These methods do not, in general, have the ability to directly report information to a caller’s context and will usually be collected as a Task result instead.

They must not call upper-half functions directly.

qemu.aqmp.util.create_task(coro: Coroutine[Any, Any, qemu.aqmp.util.T], loop: Optional[asyncio.events.AbstractEventLoop] = None) → _asyncio.Future

Python 3.6-compatible asyncio.create_task wrapper.

Parameters

• coro – The coroutine to execute in a task.

• loop – Optionally, the loop to create the task in.

Returns

An asyncio.Future object.

qemu.aqmp.util.is_closing(writer: asyncio.streams.StreamWriter) → bool

Python 3.6-compatible asyncio.StreamWriter.is_closing wrapper.

Parameters

writer – The asyncio.StreamWriter object.

Returns

True if the writer is closing, or closed.

async qemu.aqmp.util.wait_closed(writer: asyncio.streams.StreamWriter) → None

Python 3.6-compatible asyncio.StreamWriter.wait_closed wrapper.

Parameters

writer – The asyncio.StreamWriter to wait on.

qemu.aqmp.util.asyncio_run(coro: Coroutine[Any, Any, qemu.aqmp.util.T], *, debug: bool = False) → qemu.aqmp.util.T

Python 3.6-compatible asyncio.run wrapper.

Parameters

coro – A coroutine to execute now.

Returns

The return value from the coroutine.

qemu.aqmp.util.exception_summary(exc: BaseException) → str

Return a summary string of an arbitrary exception.

It will be of the form “ExceptionType: Error Message”, if the error string is non-empty, and just “ExceptionType” otherwise.

qemu.aqmp.util.pretty_traceback(prefix: str = '  | ') → str

Formats the current traceback, indented to provide visual distinction.

This is useful for printing a traceback within a traceback for debugging purposes when encapsulating errors to deliver them up the stack; when those errors are printed, this helps provide a nice visual grouping to quickly identify the parts of the error that belong to the inner exception.

Parameters

prefix – The prefix to append to each line of the traceback.

Returns

A string, formatted something like the following:

| Traceback (most recent call last):
|   File "foobar.py", line 42, in arbitrary_example
|     foo.baz()
| ArbitraryError: [Errno 42] Something bad happened!