Legacy API

(Legacy) Sync QMP Wrapper

This module provides the QEMUMonitorProtocol class, which is a synchronous wrapper around QMPClient.

Its design closely resembles that of the original QEMUMonitorProtocol class, originally written by Luiz Capitulino. It is provided here for compatibility with scripts inside the QEMU source tree that expect the old interface.

class qemu.qmp.legacy.QEMUMonitorProtocol(address: str | Tuple[str, int] | socket, server: bool = False, nickname: str | None = None)

Bases: object

Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then allow to handle commands and events.

Parameters:

• address – QEMU address, can be a unix socket path (string), a tuple in the form ( address, port ) for a TCP connection, or an existing socket.socket object.

• server – Act as the socket server. (See ‘accept’) Not applicable when passing a socket directly.

• nickname – Optional nickname used for logging.

accept(timeout: float | None = 15.0) → Dict[str, Any]

Await connection from QMP Monitor and perform capabilities negotiation.

Parameters:

timeout – timeout in seconds (nonnegative float number, or None). If None, there is no timeout, and this may block forever.

Returns:

QMP greeting dict

Raises:

ConnectError – on connection errors

clear_events() → None

Clear current list of pending events.

close() → None

Close the connection.

cmd(cmd: str, **kwds: object) → object

Build and send a QMP command to the monitor, report errors if any

cmd_obj(qmp_cmd: Dict[str, Any]) → Dict[str, Any]

Send a QMP command to the QMP Monitor.

Parameters:

qmp_cmd – QMP command to be sent as a Python dict

Returns:

QMP response as a Python dict

cmd_raw(name: str, args: Dict[str, object] | None = None) → Dict[str, Any]

Build a QMP command and send it to the QMP Monitor.

Parameters:

• name – command name (string)

• args – command arguments (dict)

connect(negotiate: bool = True) → Dict[str, Any] | None

Connect to the QMP Monitor and perform capabilities negotiation.

Returns:

QMP greeting dict, or None if negotiate is false

Raises:

ConnectError – on connection errors

get_events(wait: bool | float = False) → List[Dict[str, Any]]

Get a list of QMP events and clear all pending events.

Parameters:

wait – If False or 0, do not wait. Return None if no events ready. If True, wait until we have at least one event. Otherwise, wait for up to the specified number of seconds for at least one event.

Raises:

asyncio.TimeoutError – When a timeout is requested and the timeout period elapses.

Returns:

A list of QMP events.

classmethod parse_address(address: str) → str | Tuple[str, int]

Parse a string into a QMP address.

Figure out if the argument is in the port:host form. If it’s not, it’s probably a file path.

pull_event(wait: bool | float = False) → Dict[str, Any] | None

Pulls a single event.

Parameters:

wait – If False or 0, do not wait. Return None if no events ready. If True, wait forever until the next event. Otherwise, wait for the specified number of seconds.

Raises:

asyncio.TimeoutError – When a timeout is requested and the timeout period elapses.

Returns:

The first available QMP event, or None.

send_fd_scm(fd: int) → None

Send a file descriptor to the remote via SCM_RIGHTS.

settimeout(timeout: float | None) → None

Set the timeout for QMP RPC execution.

This timeout affects the cmd, cmd_obj, and cmd_raw methods. The accept, pull_event and get_events methods have their own configurable timeouts.

Parameters:

timeout – timeout in seconds, or None. None will wait indefinitely.

exception qemu.qmp.legacy.QMPBadPortError

Bases: QMPError

Unable to parse socket address: Port was non-numerical.

qemu.qmp.legacy.QMPMessage

QMPMessage is an entire QMP message of any kind.

alias of Dict[str, Any]

qemu.qmp.legacy.QMPObject

QMPObject is any object in a QMP message.

alias of Dict[str, object]

qemu.qmp.legacy.QMPReturnValue

QMPReturnValue is the ‘return’ value of a command.