Skip to content

Fix several typing issues, revealed after the latest mypy release. #1167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 16, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 13 additions & 11 deletions prompt_toolkit/application/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,9 @@
_AppResult = TypeVar("_AppResult")
ApplicationEventHandler = Callable[["Application[_AppResult]"], None]

_SIGWINCH = getattr(signal, "SIGWINCH", None)
_SIGTSTP = getattr(signal, "SIGTSTP", None)


class Application(Generic[_AppResult]):
"""
Expand Down Expand Up @@ -686,10 +689,9 @@ def flush_input() -> None:
self._redraw()
self._start_auto_refresh_task()

has_sigwinch = hasattr(signal, "SIGWINCH") and in_main_thread()
if has_sigwinch:
previous_winch_handler = signal.getsignal(signal.SIGWINCH)
loop.add_signal_handler(signal.SIGWINCH, self._on_resize)
if _SIGWINCH is not None and in_main_thread():
previous_winch_handler = signal.getsignal(_SIGWINCH)
loop.add_signal_handler(_SIGWINCH, self._on_resize)
if previous_winch_handler is None:
# In some situations we receive `None`. This is
# however not a valid value for passing to
Expand Down Expand Up @@ -727,9 +729,9 @@ def flush_input() -> None:
if self.input.responds_to_cpr:
await self.renderer.wait_for_cpr_responses()

if has_sigwinch:
loop.remove_signal_handler(signal.SIGWINCH)
signal.signal(signal.SIGWINCH, previous_winch_handler)
if _SIGWINCH is not None:
loop.remove_signal_handler(_SIGWINCH)
signal.signal(_SIGWINCH, previous_winch_handler)

# Wait for the run-in-terminals to terminate.
previous_run_in_terminal_f = self._running_in_terminal_f
Expand Down Expand Up @@ -995,18 +997,18 @@ def suspend_to_background(self, suspend_group: bool = True) -> None:
"""
# Only suspend when the operating system supports it.
# (Not on Windows.)
if hasattr(signal, "SIGTSTP"):
if _SIGTSTP is not None:

def run() -> None:
# Send `SIGSTP` to own process.
# Send `SIGTSTP` to own process.
# This will cause it to suspend.

# Usually we want the whole process group to be suspended. This
# handles the case when input is piped from another process.
if suspend_group:
os.kill(0, signal.SIGTSTP)
os.kill(0, _SIGTSTP)
else:
os.kill(os.getpid(), signal.SIGTSTP)
os.kill(os.getpid(), _SIGTSTP)

run_in_terminal(run)

Expand Down
4 changes: 3 additions & 1 deletion prompt_toolkit/document.py
Original file line number Diff line number Diff line change
Expand Up @@ -825,7 +825,9 @@ def find_matching_bracket_position(
"""

# Look for a match.
for A, B in "()", "[]", "{}", "<>":
for pair in "()", "[]", "{}", "<>":
A = pair[0]
B = pair[1]
if self.current_char == A:
return self.find_enclosing_bracket_right(A, B, end_pos=end_pos) or 0
elif self.current_char == B:
Expand Down
23 changes: 15 additions & 8 deletions prompt_toolkit/shortcuts/progress_bar/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import traceback
from asyncio import get_event_loop, new_event_loop, set_event_loop
from typing import (
TYPE_CHECKING,
Generic,
Iterable,
List,
Expand Down Expand Up @@ -65,6 +66,8 @@

E = KeyPressEvent

_SIGWINCH = getattr(signal, "SIGWINCH", None)


def create_key_bindings() -> KeyBindings:
"""
Expand Down Expand Up @@ -142,11 +145,15 @@ def __init__(

self._loop = get_event_loop()
self._app_loop = new_event_loop()
self._previous_winch_handler = (
signal.getsignal(signal.SIGWINCH) if hasattr(signal, "SIGWINCH") else None
)

self._previous_winch_handler = None
self._has_sigwinch = False

if TYPE_CHECKING:
# Infer type from getsignal result, as defined in typeshed. Too
# complex to repeat here.
self._previous_winch_handler = signal.getsignal(_SIGWINCH)

def __enter__(self) -> "ProgressBar":
# Create UI Application.
title_toolbar = ConditionalContainer(
Expand Down Expand Up @@ -225,10 +232,10 @@ def run() -> None:

# Attach WINCH signal handler in main thread.
# (Interrupt that we receive during resize events.)
self._has_sigwinch = hasattr(signal, "SIGWINCH") and in_main_thread()
self._has_sigwinch = _SIGWINCH is not None and in_main_thread()
if self._has_sigwinch:
self._previous_winch_handler = signal.getsignal(signal.SIGWINCH)
self._loop.add_signal_handler(signal.SIGWINCH, self.invalidate)
self._previous_winch_handler = signal.getsignal(_SIGWINCH)
self._loop.add_signal_handler(_SIGWINCH, self.invalidate)

return self

Expand All @@ -239,8 +246,8 @@ def __exit__(self, *a: object) -> None:

# Remove WINCH handler.
if self._has_sigwinch:
self._loop.remove_signal_handler(signal.SIGWINCH)
signal.signal(signal.SIGWINCH, self._previous_winch_handler)
self._loop.remove_signal_handler(_SIGWINCH)
signal.signal(_SIGWINCH, self._previous_winch_handler)

if self._thread is not None:
self._thread.join()
Expand Down