From 70aa9ca4eda4c0055f8689b04e1fef6eef0c6fc1 Mon Sep 17 00:00:00 2001 From: Jonathan Slenders Date: Tue, 12 Dec 2023 20:27:49 +0000 Subject: [PATCH] Fix type annotations in various places. --- examples/asyncio-ssh-python-embed.py | 17 +++----- ptpython/contrib/asyncssh_repl.py | 26 +++++++------ ptpython/python_input.py | 58 ++++++++++++++++++---------- ptpython/repl.py | 2 +- 4 files changed, 59 insertions(+), 44 deletions(-) diff --git a/examples/asyncio-ssh-python-embed.py b/examples/asyncio-ssh-python-embed.py index 86b56073..98d22d86 100755 --- a/examples/asyncio-ssh-python-embed.py +++ b/examples/asyncio-ssh-python-embed.py @@ -32,31 +32,24 @@ def session_requested(self): return ReplSSHServerSession(self.get_namespace) -def main(port=8222): +async def main(port: int = 8222) -> None: """ Example that starts the REPL through an SSH server. """ - loop = asyncio.get_event_loop() - # Namespace exposed in the REPL. environ = {"hello": "world"} # Start SSH server. - def create_server(): + def create_server() -> MySSHServer: return MySSHServer(lambda: environ) print("Listening on :%i" % port) print('To connect, do "ssh localhost -p %i"' % port) - loop.run_until_complete( - asyncssh.create_server( - create_server, "", port, server_host_keys=["/etc/ssh/ssh_host_dsa_key"] - ) + await asyncssh.create_server( + create_server, "", port, server_host_keys=["/tmp/ssh_host_dsa_key"] ) - # Run eventloop. - loop.run_forever() - if __name__ == "__main__": - main() + asyncio.run(main()) diff --git a/ptpython/contrib/asyncssh_repl.py b/ptpython/contrib/asyncssh_repl.py index 051519de..35da7426 100644 --- a/ptpython/contrib/asyncssh_repl.py +++ b/ptpython/contrib/asyncssh_repl.py @@ -9,20 +9,20 @@ from __future__ import annotations import asyncio -from typing import Any, TextIO, cast +from typing import Any, AnyStr, TextIO, cast import asyncssh from prompt_toolkit.data_structures import Size from prompt_toolkit.input import create_pipe_input from prompt_toolkit.output.vt100 import Vt100_Output -from ptpython.python_input import _GetNamespace +from ptpython.python_input import _GetNamespace, _Namespace from ptpython.repl import PythonRepl __all__ = ["ReplSSHServerSession"] -class ReplSSHServerSession(asyncssh.SSHServerSession): +class ReplSSHServerSession(asyncssh.SSHServerSession[str]): """ SSH server session that runs a Python REPL. @@ -35,7 +35,7 @@ def __init__( ) -> None: self._chan: Any = None - def _globals() -> dict: + def _globals() -> _Namespace: data = get_globals() data.setdefault("print", self._print) return data @@ -79,7 +79,7 @@ def _get_size(self) -> Size: width, height, pixwidth, pixheight = self._chan.get_terminal_size() return Size(rows=height, columns=width) - def connection_made(self, chan): + def connection_made(self, chan: Any) -> None: """ Client connected, run repl in coroutine. """ @@ -89,7 +89,7 @@ def connection_made(self, chan): f = asyncio.ensure_future(self.repl.run_async()) # Close channel when done. - def done(_) -> None: + def done(_: object) -> None: chan.close() self._chan = None @@ -98,24 +98,28 @@ def done(_) -> None: def shell_requested(self) -> bool: return True - def terminal_size_changed(self, width, height, pixwidth, pixheight): + def terminal_size_changed( + self, width: int, height: int, pixwidth: int, pixheight: int + ) -> None: """ When the terminal size changes, report back to CLI. """ self.repl.app._on_resize() - def data_received(self, data, datatype): + def data_received(self, data: AnyStr, datatype: int | None) -> None: """ When data is received, send to inputstream of the CLI and repaint. """ self._input_pipe.send(data) - def _print(self, *data, sep=" ", end="\n", file=None) -> None: + def _print( + self, *data: object, sep: str = " ", end: str = "\n", file: Any = None + ) -> None: """ Alternative 'print' function that prints back into the SSH channel. """ # Pop keyword-only arguments. (We cannot use the syntax from the # signature. Otherwise, Python2 will give a syntax error message when # installing.) - data = sep.join(map(str, data)) - self._chan.write(data + end) + data_as_str = sep.join(map(str, data)) + self._chan.write(data_as_str + end) diff --git a/ptpython/python_input.py b/ptpython/python_input.py index 211d36c9..14995db4 100644 --- a/ptpython/python_input.py +++ b/ptpython/python_input.py @@ -6,7 +6,7 @@ from asyncio import get_event_loop from functools import partial -from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Mapping, TypeVar +from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Mapping, TypeVar, Union from prompt_toolkit.application import Application, get_app from prompt_toolkit.auto_suggest import ( @@ -31,7 +31,7 @@ ) from prompt_toolkit.document import Document from prompt_toolkit.enums import DEFAULT_BUFFER, EditingMode -from prompt_toolkit.filters import Condition +from prompt_toolkit.filters import Condition, FilterOrBool from prompt_toolkit.formatted_text import AnyFormattedText from prompt_toolkit.history import ( FileHistory, @@ -49,8 +49,13 @@ from prompt_toolkit.key_binding.bindings.open_in_editor import ( load_open_in_editor_bindings, ) +from prompt_toolkit.key_binding.key_bindings import Binding, KeyHandlerCallable +from prompt_toolkit.key_binding.key_processor import KeyPressEvent from prompt_toolkit.key_binding.vi_state import InputMode +from prompt_toolkit.keys import Keys from prompt_toolkit.layout.containers import AnyContainer +from prompt_toolkit.layout.dimension import AnyDimension +from prompt_toolkit.layout.processors import Processor from prompt_toolkit.lexers import DynamicLexer, Lexer, SimpleLexer from prompt_toolkit.output import ColorDepth, Output from prompt_toolkit.styles import ( @@ -91,22 +96,23 @@ from typing_extensions import Protocol class _SupportsLessThan(Protocol): - # Taken from typeshed. _T is used by "sorted", which needs anything + # Taken from typeshed. _T_lt is used by "sorted", which needs anything # sortable. def __lt__(self, __other: Any) -> bool: ... -_T = TypeVar("_T", bound="_SupportsLessThan") +_T_lt = TypeVar("_T_lt", bound="_SupportsLessThan") +_T_kh = TypeVar("_T_kh", bound=Union[KeyHandlerCallable, Binding]) -class OptionCategory(Generic[_T]): - def __init__(self, title: str, options: list[Option[_T]]) -> None: +class OptionCategory(Generic[_T_lt]): + def __init__(self, title: str, options: list[Option[_T_lt]]) -> None: self.title = title self.options = options -class Option(Generic[_T]): +class Option(Generic[_T_lt]): """ Ptpython configuration option that can be shown and modified from the sidebar. @@ -122,10 +128,10 @@ def __init__( self, title: str, description: str, - get_current_value: Callable[[], _T], + get_current_value: Callable[[], _T_lt], # We accept `object` as return type for the select functions, because # often they return an unused boolean. Maybe this can be improved. - get_values: Callable[[], Mapping[_T, Callable[[], object]]], + get_values: Callable[[], Mapping[_T_lt, Callable[[], object]]], ) -> None: self.title = title self.description = description @@ -133,7 +139,7 @@ def __init__( self.get_values = get_values @property - def values(self) -> Mapping[_T, Callable[[], object]]: + def values(self) -> Mapping[_T_lt, Callable[[], object]]: return self.get_values() def activate_next(self, _previous: bool = False) -> None: @@ -208,10 +214,10 @@ def __init__( _completer: Completer | None = None, _validator: Validator | None = None, _lexer: Lexer | None = None, - _extra_buffer_processors=None, + _extra_buffer_processors: list[Processor] | None = None, _extra_layout_body: AnyContainer | None = None, - _extra_toolbars=None, - _input_buffer_height=None, + _extra_toolbars: list[AnyContainer] | None = None, + _input_buffer_height: AnyDimension | None = None, ) -> None: self.get_globals: _GetNamespace = get_globals or (lambda: {}) self.get_locals: _GetNamespace = get_locals or self.get_globals @@ -466,24 +472,36 @@ def get_compiler_flags(self) -> int: return flags - @property - def add_key_binding(self) -> Callable[[_T], _T]: + def add_key_binding( + self, + *keys: Keys | str, + filter: FilterOrBool = True, + eager: FilterOrBool = False, + is_global: FilterOrBool = False, + save_before: Callable[[KeyPressEvent], bool] = (lambda e: True), + record_in_macro: FilterOrBool = True, + ) -> Callable[[_T_kh], _T_kh]: """ Shortcut for adding new key bindings. (Mostly useful for a config.py file, that receives a PythonInput/Repl instance as input.) + All arguments are identical to prompt_toolkit's `KeyBindings.add`. + :: @python_input.add_key_binding(Keys.ControlX, filter=...) def handler(event): ... """ - - def add_binding_decorator(*k, **kw): - return self.extra_key_bindings.add(*k, **kw) - - return add_binding_decorator + return self.extra_key_bindings.add( + *keys, + filter=filter, + eager=eager, + is_global=is_global, + save_before=save_before, + record_in_macro=record_in_macro, + ) def install_code_colorscheme(self, name: str, style: BaseStyle) -> None: """ diff --git a/ptpython/repl.py b/ptpython/repl.py index 98b01afa..1db2e64f 100644 --- a/ptpython/repl.py +++ b/ptpython/repl.py @@ -158,7 +158,7 @@ def run(self) -> None: clear_title() self._remove_from_namespace() - async def run_and_show_expression_async(self, text: str): + async def run_and_show_expression_async(self, text: str) -> object: loop = asyncio.get_event_loop() try: