Skip to content

Commit

Permalink
Fix type annotations in various places.
Browse files Browse the repository at this point in the history
  • Loading branch information
jonathanslenders committed Dec 12, 2023
1 parent 68ce105 commit 70aa9ca
Show file tree
Hide file tree
Showing 4 changed files with 59 additions and 44 deletions.
17 changes: 5 additions & 12 deletions examples/asyncio-ssh-python-embed.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
26 changes: 15 additions & 11 deletions ptpython/contrib/asyncssh_repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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.
"""
Expand All @@ -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

Expand All @@ -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)
58 changes: 38 additions & 20 deletions ptpython/python_input.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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,
Expand All @@ -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 (
Expand Down Expand Up @@ -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.
Expand All @@ -122,18 +128,18 @@ 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
self.get_current_value = get_current_value
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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
"""
Expand Down
2 changes: 1 addition & 1 deletion ptpython/repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down

0 comments on commit 70aa9ca

Please sign in to comment.