From 9cf539a2a8c0055ba2a42d26677ec392e0e4f7c1 Mon Sep 17 00:00:00 2001 From: Philipp Hahn Date: Mon, 27 Apr 2020 11:10:13 +0200 Subject: examples: Add/fix PEP 484 type annotation Signed-off-by: Philipp Hahn --- examples/consolecallback.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) (limited to 'examples/consolecallback.py') diff --git a/examples/consolecallback.py b/examples/consolecallback.py index 950e312..e8820f7 100644 --- a/examples/consolecallback.py +++ b/examples/consolecallback.py @@ -2,11 +2,12 @@ # consolecallback - provide a persistent console that survives guest reboots import sys, os, logging, libvirt, tty, termios, atexit +from typing import Optional # noqa F401 -def reset_term(): +def reset_term() -> None: termios.tcsetattr(0, termios.TCSADRAIN, attrs) -def error_handler(unused, error): +def error_handler(unused, error) -> None: # The console stream errors on VM shutdown; we don't care if (error[0] == libvirt.VIR_ERR_RPC and error[1] == libvirt.VIR_FROM_STREAMS): @@ -14,20 +15,20 @@ def error_handler(unused, error): logging.warn(error) class Console(object): - def __init__(self, uri, uuid): + def __init__(self, uri: str, uuid: str) -> None: self.uri = uri self.uuid = uuid self.connection = libvirt.open(uri) self.domain = self.connection.lookupByUUIDString(uuid) self.state = self.domain.state(0) self.connection.domainEventRegister(lifecycle_callback, self) - self.stream = None + self.stream = None # type: Optional[libvirt.virStream] self.run_console = True self.stdin_watch = -1 logging.info("%s initial state %d, reason %d", self.uuid, self.state[0], self.state[1]) -def check_console(console): +def check_console(console: Console) -> bool: if (console.state[0] == libvirt.VIR_DOMAIN_RUNNING or console.state[0] == libvirt.VIR_DOMAIN_PAUSED): if console.stream is None: @@ -41,7 +42,7 @@ def check_console(console): return console.run_console -def stdin_callback(watch, fd, events, console): +def stdin_callback(watch: int, fd: int, events: int, console: Console) -> None: readbuf = os.read(fd, 1024) if readbuf.startswith(b""): console.run_console = False @@ -49,7 +50,7 @@ def stdin_callback(watch, fd, events, console): if console.stream: console.stream.send(readbuf) -def stream_callback(stream, events, console): +def stream_callback(stream: libvirt.virStream, events: int, console: Console) -> None: try: assert console.stream received_data = console.stream.recv(1024) @@ -57,7 +58,7 @@ def stream_callback(stream, events, console): return os.write(0, received_data) -def lifecycle_callback (connection, domain, event, detail, console): +def lifecycle_callback(connection: libvirt.virConnect, domain: libvirt.virDomain, event: int, detail: int, console: Console) -> None: console.state = console.domain.state(0) logging.info("%s transitioned to state %d, reason %d", console.uuid, console.state[0], console.state[1]) -- cgit v1.2.1