| Message ID | 20210730201846.5147-7-niteesh.gs@gmail.com (mailing list archive) |
|---|---|
| State | New, archived |
| Headers | show |
| Series | AQMP TUI Draft | expand |
On Fri, Jul 30, 2021 at 4:19 PM G S Niteesh Babu <niteesh.gs@gmail.com> wrote: > This patch adds type annotations for aqmp-tui using > the mypy library. > > Awesome, thanks for taking a swing at this. Looks like it wasn't as bad as I was fearing. > Signed-off-by: G S Niteesh Babu <niteesh.gs@gmail.com> > --- > python/qemu/aqmp/aqmp_tui.py | 79 ++++++++++++++++++++---------------- > python/setup.cfg | 3 -- > 2 files changed, 43 insertions(+), 39 deletions(-) > > diff --git a/python/qemu/aqmp/aqmp_tui.py b/python/qemu/aqmp/aqmp_tui.py > index ec9eba0aa7..ab9ada793a 100644 > --- a/python/qemu/aqmp/aqmp_tui.py > +++ b/python/qemu/aqmp/aqmp_tui.py > @@ -9,8 +9,15 @@ > import argparse > import asyncio > import logging > -from logging import Handler > +from logging import Handler, LogRecord > import signal > +from typing import ( > + Any, > + List, > + Optional, > + Tuple, > + Union, > +) > > import urwid > import urwid_readline > @@ -22,13 +29,13 @@ > from .util import create_task, pretty_traceback > > > -UPDATE_MSG = 'UPDATE_MSG' > +UPDATE_MSG: str = 'UPDATE_MSG' > > # Using root logger to enable all loggers under qemu and asyncio > LOGGER = logging.getLogger() > > > -def format_json(msg): > +def format_json(msg: str) -> str: > """ > Formats given multiline JSON message into a single line message. > Converting into single line is more asthetically pleasing when looking > @@ -43,17 +50,17 @@ def format_json(msg): > > > class App(QMPClient): > - def __init__(self, address): > + def __init__(self, address: Union[str, Tuple[str, int]]) -> None: > urwid.register_signal(type(self), UPDATE_MSG) > self.window = Window(self) > self.address = address > - self.aloop = None > + self.aloop: Optional[Any] = None # FIXME: Use more concrete type. > I ran into this in util.py; you want Optional[asyncio.AbstractEventLoop] here. > super().__init__() > > - def add_to_history(self, msg): > + def add_to_history(self, msg: str) -> None: > urwid.emit_signal(self, UPDATE_MSG, msg) > > - def _cb_outbound(self, msg): > + def _cb_outbound(self, msg: Message) -> Message: > # FIXME: I think the ideal way to omit these messages during > in-TUI > # logging will be to add a filter to the logger. We can use regex > to > # filter out messages starting with 'Request:' or 'Response:' but > I > @@ -67,25 +74,25 @@ def _cb_outbound(self, msg): > self.add_to_history('<-- ' + str(msg)) > return msg > > - def _cb_inbound(self, msg): > + def _cb_inbound(self, msg: Message) -> Message: > handler = LOGGER.handlers[0] > if not isinstance(handler, TUILogHandler): > LOGGER.debug('Response: %s', str(msg)) > self.add_to_history('--> ' + str(msg)) > return msg > > - async def wait_for_events(self): > + async def wait_for_events(self) -> None: > async for event in self.events: > self.handle_event(event) > > - async def _send_to_server(self, raw_msg): > + async def _send_to_server(self, raw_msg: str) -> None: > # FIXME: Format the raw_msg in history view to one line. It is not > # pleasing to see multiple lines JSON object with an error > statement. > try: > msg = Message(bytes(raw_msg, encoding='utf-8')) > # Format multiline json into a single line JSON, since it is > more > # pleasing to look along with err message in TUI. > - raw_msg = self.format_json(raw_msg) > + raw_msg = format_json(raw_msg) > await self._raw(msg, assign_id='id' not in msg) > except (ValueError, TypeError) as err: > LOGGER.info('Invalid message: %s', str(err)) > @@ -102,18 +109,18 @@ def _cb_inbound(self, msg): > LOGGER.error('Exception from _send_to_server: %s', str(err)) > raise err > > - def cb_send_to_server(self, msg): > + def cb_send_to_server(self, msg: str) -> None: > create_task(self._send_to_server(msg)) > > - def unhandled_input(self, key): > + def unhandled_input(self, key: str) -> None: > if key == 'esc': > self.kill_app() > > - def kill_app(self): > + def kill_app(self) -> None: > # TODO: Work on the disconnect logic > create_task(self._kill_app()) > > - async def _kill_app(self): > + async def _kill_app(self) -> None: > # It is ok to call disconnect even in disconnect state > try: > await self.disconnect() > @@ -124,7 +131,7 @@ def kill_app(self): > raise err > raise urwid.ExitMainLoop() > > - def handle_event(self, event): > + def handle_event(self, event: Message) -> None: > # FIXME: Consider all states present in qapi/run-state.json > if event['event'] == 'SHUTDOWN': > self._set_status('Server shutdown') > @@ -139,7 +146,7 @@ def _get_formatted_address(self) -> str: > addr = f'{host}:{port}' > return addr > > - async def connect_server(self): > + async def connect_server(self) -> None: > try: > await self.connect(self.address) > addr = self._get_formatted_address() > @@ -148,7 +155,7 @@ def _get_formatted_address(self) -> str: > LOGGER.info('connect_server: ConnectError %s', str(err)) > self._set_status('Server shutdown') > > - def run(self, debug=False): > + def run(self, debug: bool = False) -> None: > self.aloop = asyncio.get_event_loop() > self.aloop.set_debug(debug) > > @@ -176,7 +183,7 @@ class StatusBar(urwid.Text): > """ > A simple Text widget that currently only shows connection status. > """ > - def __init__(self, text=''): > + def __init__(self, text: str = ''): > super().__init__(text, align='right') > > > @@ -185,14 +192,14 @@ class Editor(urwid_readline.ReadlineEdit): > Support urwid_readline features along with > history support which lacks in urwid_readline > """ > - def __init__(self, master): > + def __init__(self, master: App) -> None: > super().__init__(caption='> ', multiline=True) > self.master = master > - self.history = [] > - self.last_index = -1 > - self.show_history = False > + self.history: List[str] = [] > + self.last_index: int = -1 > + self.show_history: bool = False > > - def keypress(self, size, key): > + def keypress(self, size: Tuple[int, int], key: str) -> Optional[str]: > # TODO: Add some logic for down key and clean up logic if > possible. > # Returning None means the key has been handled by this widget > # which otherwise is propogated to the parent widget to be > @@ -223,7 +230,7 @@ def keypress(self, size, key): > else: > self.show_history = False > self.last_index = 0 > - return super().keypress(size, key) > + return super().keypress(size, key) # type: ignore > try using cast(Optional[str], super().keypress(size, key)) instead of the type: ignore. It doesn't make a gigantic difference, really, but it looks "narrower" in scope and will probably confuse *me* less in the future. > return None > > > @@ -231,7 +238,7 @@ class EditorWidget(urwid.Filler): > """ > Wraps CustomEdit > """ > - def __init__(self, master): > + def __init__(self, master: App) -> None: > super().__init__(Editor(master), valign='top') > > > @@ -239,12 +246,12 @@ class HistoryBox(urwid.ListBox): > """ > Shows all the QMP message transmitted/received > """ > - def __init__(self, master): > + def __init__(self, master: App) -> None: > self.master = master > self.history = urwid.SimpleFocusListWalker([]) > super().__init__(self.history) > > - def add_to_history(self, history): > + def add_to_history(self, history: str) -> None: > self.history.append(urwid.Text(history)) > if self.history: > self.history.set_focus(len(self.history) - 1) > @@ -254,7 +261,7 @@ class HistoryWindow(urwid.Frame): > """ > Composes the HistoryBox and EditorWidget > """ > - def __init__(self, master): > + def __init__(self, master: App) -> None: > self.master = master > self.editor_widget = EditorWidget(master) > self.editor = urwid.LineBox(self.editor_widget) > @@ -264,7 +271,7 @@ def __init__(self, master): > super().__init__(self.body) > urwid.connect_signal(self.master, UPDATE_MSG, > self.cb_add_to_history) > > - def cb_add_to_history(self, msg): > + def cb_add_to_history(self, msg: str) -> None: > self.history.add_to_history(msg) > > > @@ -275,7 +282,7 @@ class Window(urwid.Frame): > future when we will have multiple windows and want to the switch > between > them and display overlays > """ > - def __init__(self, master): > + def __init__(self, master: App) -> None: > self.master = master > footer = StatusBar() > body = HistoryWindow(master) > @@ -283,18 +290,18 @@ def __init__(self, master): > > > class TUILogHandler(Handler): > - def __init__(self, tui): > + def __init__(self, tui: App) -> None: > super().__init__() > self.tui = tui > > - def emit(self, record): > + def emit(self, record: LogRecord) -> None: > level = record.levelname > msg = record.getMessage() > msg = f'[{level}]: {msg}' > self.tui.add_to_history(msg) > > > -def main(): > +def main() -> None: > parser = argparse.ArgumentParser(description='AQMP TUI') > parser.add_argument('qmp_server', help='Address of the QMP server' > '< UNIX socket path | TCP addr:port >') > @@ -311,7 +318,7 @@ def main(): > try: > address = QEMUMonitorProtocol.parse_address(args.qmp_server) > except QMPBadPortError as err: > - parser.error(err) > + parser.error(str(err)) > > app = App(address) > > @@ -330,4 +337,4 @@ def main(): > > > if __name__ == '__main__': > - main() # type: ignore > + main() > diff --git a/python/setup.cfg b/python/setup.cfg > index 8cd9ac0d81..11c6240aba 100644 > --- a/python/setup.cfg > +++ b/python/setup.cfg > @@ -83,9 +83,6 @@ namespace_packages = True > allow_subclassing_any = True > > [mypy-qemu.aqmp.aqmp_tui] > -disallow_untyped_defs = False > -disallow_incomplete_defs = False > -check_untyped_defs = False > # urwid and urwid_readline have no type stubs: > allow_subclassing_any = True > > -- > 2.17.1 > > Please squash this into the first patch, but it looks good, thanks! --js
On Fri, Aug 6, 2021 at 1:26 AM John Snow <jsnow@redhat.com> wrote: > > > On Fri, Jul 30, 2021 at 4:19 PM G S Niteesh Babu <niteesh.gs@gmail.com> > wrote: > >> This patch adds type annotations for aqmp-tui using >> the mypy library. >> >> > Awesome, thanks for taking a swing at this. Looks like it wasn't as bad as > I was fearing. > > >> Signed-off-by: G S Niteesh Babu <niteesh.gs@gmail.com> >> --- >> python/qemu/aqmp/aqmp_tui.py | 79 ++++++++++++++++++++---------------- >> python/setup.cfg | 3 -- >> 2 files changed, 43 insertions(+), 39 deletions(-) >> >> diff --git a/python/qemu/aqmp/aqmp_tui.py b/python/qemu/aqmp/aqmp_tui.py >> index ec9eba0aa7..ab9ada793a 100644 >> --- a/python/qemu/aqmp/aqmp_tui.py >> +++ b/python/qemu/aqmp/aqmp_tui.py >> @@ -9,8 +9,15 @@ >> import argparse >> import asyncio >> import logging >> -from logging import Handler >> +from logging import Handler, LogRecord >> import signal >> +from typing import ( >> + Any, >> + List, >> + Optional, >> + Tuple, >> + Union, >> +) >> >> import urwid >> import urwid_readline >> @@ -22,13 +29,13 @@ >> from .util import create_task, pretty_traceback >> >> >> -UPDATE_MSG = 'UPDATE_MSG' >> +UPDATE_MSG: str = 'UPDATE_MSG' >> >> # Using root logger to enable all loggers under qemu and asyncio >> LOGGER = logging.getLogger() >> >> >> -def format_json(msg): >> +def format_json(msg: str) -> str: >> """ >> Formats given multiline JSON message into a single line message. >> Converting into single line is more asthetically pleasing when >> looking >> @@ -43,17 +50,17 @@ def format_json(msg): >> >> >> class App(QMPClient): >> - def __init__(self, address): >> + def __init__(self, address: Union[str, Tuple[str, int]]) -> None: >> urwid.register_signal(type(self), UPDATE_MSG) >> self.window = Window(self) >> self.address = address >> - self.aloop = None >> + self.aloop: Optional[Any] = None # FIXME: Use more concrete >> type. >> > > I ran into this in util.py; you want Optional[asyncio.AbstractEventLoop] > here. > Thanks. Fixed. > > >> super().__init__() >> >> - def add_to_history(self, msg): >> + def add_to_history(self, msg: str) -> None: >> urwid.emit_signal(self, UPDATE_MSG, msg) >> >> - def _cb_outbound(self, msg): >> + def _cb_outbound(self, msg: Message) -> Message: >> # FIXME: I think the ideal way to omit these messages during >> in-TUI >> # logging will be to add a filter to the logger. We can use >> regex to >> # filter out messages starting with 'Request:' or 'Response:' >> but I >> @@ -67,25 +74,25 @@ def _cb_outbound(self, msg): >> self.add_to_history('<-- ' + str(msg)) >> return msg >> >> - def _cb_inbound(self, msg): >> + def _cb_inbound(self, msg: Message) -> Message: >> handler = LOGGER.handlers[0] >> if not isinstance(handler, TUILogHandler): >> LOGGER.debug('Response: %s', str(msg)) >> self.add_to_history('--> ' + str(msg)) >> return msg >> >> - async def wait_for_events(self): >> + async def wait_for_events(self) -> None: >> async for event in self.events: >> self.handle_event(event) >> >> - async def _send_to_server(self, raw_msg): >> + async def _send_to_server(self, raw_msg: str) -> None: >> # FIXME: Format the raw_msg in history view to one line. It is >> not >> # pleasing to see multiple lines JSON object with an error >> statement. >> try: >> msg = Message(bytes(raw_msg, encoding='utf-8')) >> # Format multiline json into a single line JSON, since it is >> more >> # pleasing to look along with err message in TUI. >> - raw_msg = self.format_json(raw_msg) >> + raw_msg = format_json(raw_msg) >> await self._raw(msg, assign_id='id' not in msg) >> except (ValueError, TypeError) as err: >> LOGGER.info('Invalid message: %s', str(err)) >> @@ -102,18 +109,18 @@ def _cb_inbound(self, msg): >> LOGGER.error('Exception from _send_to_server: %s', str(err)) >> raise err >> >> - def cb_send_to_server(self, msg): >> + def cb_send_to_server(self, msg: str) -> None: >> create_task(self._send_to_server(msg)) >> >> - def unhandled_input(self, key): >> + def unhandled_input(self, key: str) -> None: >> if key == 'esc': >> self.kill_app() >> >> - def kill_app(self): >> + def kill_app(self) -> None: >> # TODO: Work on the disconnect logic >> create_task(self._kill_app()) >> >> - async def _kill_app(self): >> + async def _kill_app(self) -> None: >> # It is ok to call disconnect even in disconnect state >> try: >> await self.disconnect() >> @@ -124,7 +131,7 @@ def kill_app(self): >> raise err >> raise urwid.ExitMainLoop() >> >> - def handle_event(self, event): >> + def handle_event(self, event: Message) -> None: >> # FIXME: Consider all states present in qapi/run-state.json >> if event['event'] == 'SHUTDOWN': >> self._set_status('Server shutdown') >> @@ -139,7 +146,7 @@ def _get_formatted_address(self) -> str: >> addr = f'{host}:{port}' >> return addr >> >> - async def connect_server(self): >> + async def connect_server(self) -> None: >> try: >> await self.connect(self.address) >> addr = self._get_formatted_address() >> @@ -148,7 +155,7 @@ def _get_formatted_address(self) -> str: >> LOGGER.info('connect_server: ConnectError %s', str(err)) >> self._set_status('Server shutdown') >> >> - def run(self, debug=False): >> + def run(self, debug: bool = False) -> None: >> self.aloop = asyncio.get_event_loop() >> self.aloop.set_debug(debug) >> >> @@ -176,7 +183,7 @@ class StatusBar(urwid.Text): >> """ >> A simple Text widget that currently only shows connection status. >> """ >> - def __init__(self, text=''): >> + def __init__(self, text: str = ''): >> super().__init__(text, align='right') >> >> >> @@ -185,14 +192,14 @@ class Editor(urwid_readline.ReadlineEdit): >> Support urwid_readline features along with >> history support which lacks in urwid_readline >> """ >> - def __init__(self, master): >> + def __init__(self, master: App) -> None: >> super().__init__(caption='> ', multiline=True) >> self.master = master >> - self.history = [] >> - self.last_index = -1 >> - self.show_history = False >> + self.history: List[str] = [] >> + self.last_index: int = -1 >> + self.show_history: bool = False >> >> - def keypress(self, size, key): >> + def keypress(self, size: Tuple[int, int], key: str) -> Optional[str]: >> # TODO: Add some logic for down key and clean up logic if >> possible. >> # Returning None means the key has been handled by this widget >> # which otherwise is propogated to the parent widget to be >> @@ -223,7 +230,7 @@ def keypress(self, size, key): >> else: >> self.show_history = False >> self.last_index = 0 >> - return super().keypress(size, key) >> + return super().keypress(size, key) # type: ignore >> > > try using cast(Optional[str], super().keypress(size, key)) instead of the > type: ignore. It doesn't make a gigantic difference, really, but it looks > "narrower" in scope and will probably confuse *me* less in the future. > Fixed > > >> return None >> >> >> @@ -231,7 +238,7 @@ class EditorWidget(urwid.Filler): >> """ >> Wraps CustomEdit >> """ >> - def __init__(self, master): >> + def __init__(self, master: App) -> None: >> super().__init__(Editor(master), valign='top') >> >> >> @@ -239,12 +246,12 @@ class HistoryBox(urwid.ListBox): >> """ >> Shows all the QMP message transmitted/received >> """ >> - def __init__(self, master): >> + def __init__(self, master: App) -> None: >> self.master = master >> self.history = urwid.SimpleFocusListWalker([]) >> super().__init__(self.history) >> >> - def add_to_history(self, history): >> + def add_to_history(self, history: str) -> None: >> self.history.append(urwid.Text(history)) >> if self.history: >> self.history.set_focus(len(self.history) - 1) >> @@ -254,7 +261,7 @@ class HistoryWindow(urwid.Frame): >> """ >> Composes the HistoryBox and EditorWidget >> """ >> - def __init__(self, master): >> + def __init__(self, master: App) -> None: >> self.master = master >> self.editor_widget = EditorWidget(master) >> self.editor = urwid.LineBox(self.editor_widget) >> @@ -264,7 +271,7 @@ def __init__(self, master): >> super().__init__(self.body) >> urwid.connect_signal(self.master, UPDATE_MSG, >> self.cb_add_to_history) >> >> - def cb_add_to_history(self, msg): >> + def cb_add_to_history(self, msg: str) -> None: >> self.history.add_to_history(msg) >> >> >> @@ -275,7 +282,7 @@ class Window(urwid.Frame): >> future when we will have multiple windows and want to the switch >> between >> them and display overlays >> """ >> - def __init__(self, master): >> + def __init__(self, master: App) -> None: >> self.master = master >> footer = StatusBar() >> body = HistoryWindow(master) >> @@ -283,18 +290,18 @@ def __init__(self, master): >> >> >> class TUILogHandler(Handler): >> - def __init__(self, tui): >> + def __init__(self, tui: App) -> None: >> super().__init__() >> self.tui = tui >> >> - def emit(self, record): >> + def emit(self, record: LogRecord) -> None: >> level = record.levelname >> msg = record.getMessage() >> msg = f'[{level}]: {msg}' >> self.tui.add_to_history(msg) >> >> >> -def main(): >> +def main() -> None: >> parser = argparse.ArgumentParser(description='AQMP TUI') >> parser.add_argument('qmp_server', help='Address of the QMP server' >> '< UNIX socket path | TCP addr:port >') >> @@ -311,7 +318,7 @@ def main(): >> try: >> address = QEMUMonitorProtocol.parse_address(args.qmp_server) >> except QMPBadPortError as err: >> - parser.error(err) >> + parser.error(str(err)) >> >> app = App(address) >> >> @@ -330,4 +337,4 @@ def main(): >> >> >> if __name__ == '__main__': >> - main() # type: ignore >> + main() >> diff --git a/python/setup.cfg b/python/setup.cfg >> index 8cd9ac0d81..11c6240aba 100644 >> --- a/python/setup.cfg >> +++ b/python/setup.cfg >> @@ -83,9 +83,6 @@ namespace_packages = True >> allow_subclassing_any = True >> >> [mypy-qemu.aqmp.aqmp_tui] >> -disallow_untyped_defs = False >> -disallow_incomplete_defs = False >> -check_untyped_defs = False >> # urwid and urwid_readline have no type stubs: >> allow_subclassing_any = True >> >> -- >> 2.17.1 >> >> > Please squash this into the first patch, but it looks good, thanks! > Thanks. > > --js >
diff --git a/python/qemu/aqmp/aqmp_tui.py b/python/qemu/aqmp/aqmp_tui.py index ec9eba0aa7..ab9ada793a 100644 --- a/python/qemu/aqmp/aqmp_tui.py +++ b/python/qemu/aqmp/aqmp_tui.py @@ -9,8 +9,15 @@ import argparse import asyncio import logging -from logging import Handler +from logging import Handler, LogRecord import signal +from typing import ( + Any, + List, + Optional, + Tuple, + Union, +) import urwid import urwid_readline @@ -22,13 +29,13 @@ from .util import create_task, pretty_traceback -UPDATE_MSG = 'UPDATE_MSG' +UPDATE_MSG: str = 'UPDATE_MSG' # Using root logger to enable all loggers under qemu and asyncio LOGGER = logging.getLogger() -def format_json(msg): +def format_json(msg: str) -> str: """ Formats given multiline JSON message into a single line message. Converting into single line is more asthetically pleasing when looking @@ -43,17 +50,17 @@ def format_json(msg): class App(QMPClient): - def __init__(self, address): + def __init__(self, address: Union[str, Tuple[str, int]]) -> None: urwid.register_signal(type(self), UPDATE_MSG) self.window = Window(self) self.address = address - self.aloop = None + self.aloop: Optional[Any] = None # FIXME: Use more concrete type. super().__init__() - def add_to_history(self, msg): + def add_to_history(self, msg: str) -> None: urwid.emit_signal(self, UPDATE_MSG, msg) - def _cb_outbound(self, msg): + def _cb_outbound(self, msg: Message) -> Message: # FIXME: I think the ideal way to omit these messages during in-TUI # logging will be to add a filter to the logger. We can use regex to # filter out messages starting with 'Request:' or 'Response:' but I @@ -67,25 +74,25 @@ def _cb_outbound(self, msg): self.add_to_history('<-- ' + str(msg)) return msg - def _cb_inbound(self, msg): + def _cb_inbound(self, msg: Message) -> Message: handler = LOGGER.handlers[0] if not isinstance(handler, TUILogHandler): LOGGER.debug('Response: %s', str(msg)) self.add_to_history('--> ' + str(msg)) return msg - async def wait_for_events(self): + async def wait_for_events(self) -> None: async for event in self.events: self.handle_event(event) - async def _send_to_server(self, raw_msg): + async def _send_to_server(self, raw_msg: str) -> None: # FIXME: Format the raw_msg in history view to one line. It is not # pleasing to see multiple lines JSON object with an error statement. try: msg = Message(bytes(raw_msg, encoding='utf-8')) # Format multiline json into a single line JSON, since it is more # pleasing to look along with err message in TUI. - raw_msg = self.format_json(raw_msg) + raw_msg = format_json(raw_msg) await self._raw(msg, assign_id='id' not in msg) except (ValueError, TypeError) as err: LOGGER.info('Invalid message: %s', str(err)) @@ -102,18 +109,18 @@ def _cb_inbound(self, msg): LOGGER.error('Exception from _send_to_server: %s', str(err)) raise err - def cb_send_to_server(self, msg): + def cb_send_to_server(self, msg: str) -> None: create_task(self._send_to_server(msg)) - def unhandled_input(self, key): + def unhandled_input(self, key: str) -> None: if key == 'esc': self.kill_app() - def kill_app(self): + def kill_app(self) -> None: # TODO: Work on the disconnect logic create_task(self._kill_app()) - async def _kill_app(self): + async def _kill_app(self) -> None: # It is ok to call disconnect even in disconnect state try: await self.disconnect() @@ -124,7 +131,7 @@ def kill_app(self): raise err raise urwid.ExitMainLoop() - def handle_event(self, event): + def handle_event(self, event: Message) -> None: # FIXME: Consider all states present in qapi/run-state.json if event['event'] == 'SHUTDOWN': self._set_status('Server shutdown') @@ -139,7 +146,7 @@ def _get_formatted_address(self) -> str: addr = f'{host}:{port}' return addr - async def connect_server(self): + async def connect_server(self) -> None: try: await self.connect(self.address) addr = self._get_formatted_address() @@ -148,7 +155,7 @@ def _get_formatted_address(self) -> str: LOGGER.info('connect_server: ConnectError %s', str(err)) self._set_status('Server shutdown') - def run(self, debug=False): + def run(self, debug: bool = False) -> None: self.aloop = asyncio.get_event_loop() self.aloop.set_debug(debug) @@ -176,7 +183,7 @@ class StatusBar(urwid.Text): """ A simple Text widget that currently only shows connection status. """ - def __init__(self, text=''): + def __init__(self, text: str = ''): super().__init__(text, align='right') @@ -185,14 +192,14 @@ class Editor(urwid_readline.ReadlineEdit): Support urwid_readline features along with history support which lacks in urwid_readline """ - def __init__(self, master): + def __init__(self, master: App) -> None: super().__init__(caption='> ', multiline=True) self.master = master - self.history = [] - self.last_index = -1 - self.show_history = False + self.history: List[str] = [] + self.last_index: int = -1 + self.show_history: bool = False - def keypress(self, size, key): + def keypress(self, size: Tuple[int, int], key: str) -> Optional[str]: # TODO: Add some logic for down key and clean up logic if possible. # Returning None means the key has been handled by this widget # which otherwise is propogated to the parent widget to be @@ -223,7 +230,7 @@ def keypress(self, size, key): else: self.show_history = False self.last_index = 0 - return super().keypress(size, key) + return super().keypress(size, key) # type: ignore return None @@ -231,7 +238,7 @@ class EditorWidget(urwid.Filler): """ Wraps CustomEdit """ - def __init__(self, master): + def __init__(self, master: App) -> None: super().__init__(Editor(master), valign='top') @@ -239,12 +246,12 @@ class HistoryBox(urwid.ListBox): """ Shows all the QMP message transmitted/received """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master self.history = urwid.SimpleFocusListWalker([]) super().__init__(self.history) - def add_to_history(self, history): + def add_to_history(self, history: str) -> None: self.history.append(urwid.Text(history)) if self.history: self.history.set_focus(len(self.history) - 1) @@ -254,7 +261,7 @@ class HistoryWindow(urwid.Frame): """ Composes the HistoryBox and EditorWidget """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master self.editor_widget = EditorWidget(master) self.editor = urwid.LineBox(self.editor_widget) @@ -264,7 +271,7 @@ def __init__(self, master): super().__init__(self.body) urwid.connect_signal(self.master, UPDATE_MSG, self.cb_add_to_history) - def cb_add_to_history(self, msg): + def cb_add_to_history(self, msg: str) -> None: self.history.add_to_history(msg) @@ -275,7 +282,7 @@ class Window(urwid.Frame): future when we will have multiple windows and want to the switch between them and display overlays """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master footer = StatusBar() body = HistoryWindow(master) @@ -283,18 +290,18 @@ def __init__(self, master): class TUILogHandler(Handler): - def __init__(self, tui): + def __init__(self, tui: App) -> None: super().__init__() self.tui = tui - def emit(self, record): + def emit(self, record: LogRecord) -> None: level = record.levelname msg = record.getMessage() msg = f'[{level}]: {msg}' self.tui.add_to_history(msg) -def main(): +def main() -> None: parser = argparse.ArgumentParser(description='AQMP TUI') parser.add_argument('qmp_server', help='Address of the QMP server' '< UNIX socket path | TCP addr:port >') @@ -311,7 +318,7 @@ def main(): try: address = QEMUMonitorProtocol.parse_address(args.qmp_server) except QMPBadPortError as err: - parser.error(err) + parser.error(str(err)) app = App(address) @@ -330,4 +337,4 @@ def main(): if __name__ == '__main__': - main() # type: ignore + main() diff --git a/python/setup.cfg b/python/setup.cfg index 8cd9ac0d81..11c6240aba 100644 --- a/python/setup.cfg +++ b/python/setup.cfg @@ -83,9 +83,6 @@ namespace_packages = True allow_subclassing_any = True [mypy-qemu.aqmp.aqmp_tui] -disallow_untyped_defs = False -disallow_incomplete_defs = False -check_untyped_defs = False # urwid and urwid_readline have no type stubs: allow_subclassing_any = True
This patch adds type annotations for aqmp-tui using the mypy library. Signed-off-by: G S Niteesh Babu <niteesh.gs@gmail.com> --- python/qemu/aqmp/aqmp_tui.py | 79 ++++++++++++++++++++---------------- python/setup.cfg | 3 -- 2 files changed, 43 insertions(+), 39 deletions(-)