dbcli/ptpython
1
0
Fork 0
Code Issues Activity

Merging upstream version 3.0.19.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-09 18:26:46 +01:00
parent df46317709
commit 03eba84c1a
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
14 changed files with 272 additions and 190 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.github/workflows/test.yaml vendored
View file

@ -23,6 +23,7 @@ jobs:
sudo apt remove python3-pip
python -m pip install --upgrade pip
python -m pip install . black isort mypy pytest readme_renderer
python -m pip install . types-dataclasses # Needed for Python 3.6
pip list
- name: Type Checker
run: |

25
CHANGELOG
View file

@ -1,6 +1,31 @@
CHANGELOG
=========
3.0.19: 2020-07-08
------------------
Fixes:
- Fix handling of `SystemExit` (fixes "ValueError: I/O operation on closed
file").
- Allow usage of `await` in assignment expressions or for-loops.
3.0.18: 2020-06-26
------------------
Fixes:
- Made "black" an optional dependency.
3.0.17: 2020-03-22
------------------
Fixes:
- Fix leaking file descriptors due to not closing the asyncio event loop after
reading input in a thread.
- Fix race condition during retrieval of signatures.
3.0.16: 2020-02-11
------------------

2
examples/ptpython_config/config.py
View file

@ -157,7 +157,7 @@ def configure(repl):
@repl.add_key_binding("j", "j", filter=ViInsertMode())
def _(event):
" Map 'jj' to Escape. "
event.cli.key_processor.feed(KeyPress("escape"))
event.cli.key_processor.feed(KeyPress(Keys("escape")))
"""
# Custom key binding for some simple autocorrection while typing.

2
ptpython/entry_points/run_ptipython.py
View file

@ -31,7 +31,7 @@ def run(user_ns=None):
path = a.args[0]
with open(path, "rb") as f:
code = compile(f.read(), path, "exec")
exec(code, {})
exec(code, {"__name__": "__main__", "__file__": path})
else:
enable_deprecation_warnings()

8
ptpython/entry_points/run_ptpython.py
View file

@ -179,9 +179,11 @@ def run() -> None:
path = a.args[0]
with open(path, "rb") as f:
code = compile(f.read(), path, "exec")
# NOTE: We have to pass an empty dictionary as namespace. Omitting
# this argument causes imports to not be found. See issue #326.
exec(code, {})
# NOTE: We have to pass a dict as namespace. Omitting this argument
# causes imports to not be found. See issue #326.
# However, an empty dict sets __name__ to 'builtins', which
# breaks `if __name__ == '__main__'` checks. See issue #444.
exec(code, {"__name__": "__main__", "__file__": path})
# Run interactive shell.
else:

17
ptpython/ipython.py
View file

@ -282,4 +282,21 @@ def embed(**kwargs):
kwargs["config"] = config
shell = InteractiveShellEmbed.instance(**kwargs)
initialize_extensions(shell, config["InteractiveShellApp"]["extensions"])
run_startup_scripts(shell)
shell(header=header, stack_depth=2, compile_flags=compile_flags)
def run_startup_scripts(shell):
"""
Contributed by linyuxu:
https://github.com/prompt-toolkit/ptpython/issues/126#issue-161242480
"""
import glob
import os
startup_dir = shell.profile_dir.startup_dir
startup_files = []
startup_files += glob.glob(os.path.join(startup_dir, "*.py"))
startup_files += glob.glob(os.path.join(startup_dir, "*.ipy"))
for file in startup_files:
shell.run_cell(open(file).read())

4
ptpython/layout.py
View file

@ -646,7 +646,7 @@ class PtPythonLayout:
sidebar = python_sidebar(python_input)
self.exit_confirmation = create_exit_confirmation(python_input)
root_container = HSplit(
self.root_container = HSplit(
[
VSplit(
[
@ -759,5 +759,5 @@ class PtPythonLayout:
]
)
self.layout = Layout(root_container)
self.layout = Layout(self.root_container)
self.sidebar = sidebar

98
ptpython/python_input.py
View file

@ -4,7 +4,6 @@ This can be used for creation of Python REPLs.
"""
import __future__
import threading
from asyncio import get_event_loop
from functools import partial
from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, List, Optional, TypeVar
@ -174,6 +173,11 @@ class PythonInput:
python_input = PythonInput(...)
python_code = python_input.app.run()
:param create_app: When `False`, don't create and manage a prompt_toolkit
application. The default is `True` and should only be set
to false if PythonInput is being embedded in a separate
prompt_toolkit application.
"""
def __init__(
@ -188,6 +192,7 @@ class PythonInput:
output: Optional[Output] = None,
# For internal use.
extra_key_bindings: Optional[KeyBindings] = None,
create_app=True,
_completer: Optional[Completer] = None,
_validator: Optional[Validator] = None,
_lexer: Optional[Lexer] = None,
@ -380,10 +385,16 @@ class PythonInput:
extra_toolbars=self._extra_toolbars,
)
self.app = self._create_application(input, output)
# Create an app if requested. If not, the global get_app() is returned
# for self.app via property getter.
if create_app:
self._app: Optional[Application] = self._create_application(input, output)
# Setting vi_mode will not work unless the prompt_toolkit
# application has been created.
if vi_mode:
self.app.editing_mode = EditingMode.VI
else:
self._app = None
def _accept_handler(self, buff: Buffer) -> bool:
app = get_app()
@ -914,23 +925,19 @@ class PythonInput:
else:
self.editing_mode = EditingMode.EMACS
def _on_input_timeout(self, buff: Buffer, loop=None) -> None:
@property
def app(self) -> Application:
if self._app is None:
return get_app()
return self._app
def _on_input_timeout(self, buff: Buffer) -> None:
"""
When there is no input activity,
in another thread, get the signature of the current code.
"""
app = self.app
# Never run multiple get-signature threads.
if self._get_signatures_thread_running:
return
self._get_signatures_thread_running = True
document = buff.document
loop = loop or get_event_loop()
def run():
def get_signatures_in_executor(document: Document) -> List[Signature]:
# First, get signatures from Jedi. If we didn't found any and if
# "dictionary completion" (eval-based completion) is enabled, then
# get signatures using eval.
@ -942,11 +949,33 @@ class PythonInput:
document, self.get_locals(), self.get_globals()
)
return signatures
app = self.app
async def on_timeout_task() -> None:
loop = get_event_loop()
# Never run multiple get-signature threads.
if self._get_signatures_thread_running:
return
self._get_signatures_thread_running = True
try:
while True:
document = buff.document
signatures = await loop.run_in_executor(
None, get_signatures_in_executor, document
)
# If the text didn't change in the meantime, take these
# signatures. Otherwise, try again.
if buff.text == document.text:
break
finally:
self._get_signatures_thread_running = False
# Set signatures and redraw if the text didn't change in the
# meantime. Otherwise request new signatures.
if buff.text == document.text:
# Set signatures and redraw.
self.signatures = signatures
# Set docstring in docstring buffer.
@ -958,10 +987,9 @@ class PythonInput:
self.docstring_buffer.reset()
app.invalidate()
else:
self._on_input_timeout(buff, loop=loop)
loop.run_in_executor(None, run)
if app.is_running:
app.create_background_task(on_timeout_task())
def on_reset(self) -> None:
self.signatures = []
@ -970,7 +998,7 @@ class PythonInput:
"""
Display the history.
"""
app = get_app()
app = self.app
app.vi_state.input_mode = InputMode.NAVIGATION
history = PythonHistory(self, self.default_buffer.document)
@ -1012,15 +1040,9 @@ class PythonInput:
self.app.vi_state.input_mode = InputMode.NAVIGATION
# Run the UI.
result: str = ""
exception: Optional[BaseException] = None
def in_thread() -> None:
nonlocal result, exception
try:
while True:
try:
result = self.app.run(pre_run=pre_run)
result = self.app.run(pre_run=pre_run, in_thread=True)
if result.lstrip().startswith("\x1a"):
# When the input starts with Ctrl-Z, quit the REPL.
@ -1033,22 +1055,10 @@ class PythonInput:
result = unindent_code(result)
if result and not result.isspace():
return
except KeyboardInterrupt:
# Abort - try again.
self.default_buffer.document = Document()
except BaseException as e:
exception = e
return
finally:
if self.insert_blank_line_after_input:
self.app.output.write("\n")
thread = threading.Thread(target=in_thread)
thread.start()
thread.join()
if exception is not None:
raise exception
return result
except KeyboardInterrupt:
# Abort - try again.
self.default_buffer.document = Document()

151
ptpython/repl.py
View file

@ -11,7 +11,6 @@ import asyncio
import builtins
import os
import sys
import threading
import traceback
import types
import warnings
@ -91,31 +90,15 @@ class PythonRepl(PythonInput):
output = self.app.output
output.write("WARNING | File not found: {}\n\n".format(path))
def run(self) -> None:
"""
Run the REPL loop.
"""
if self.terminal_title:
set_title(self.terminal_title)
self._add_to_namespace()
def run_and_show_expression(self, expression):
try:
while True:
try:
# Read.
try:
text = self.read()
except EOFError:
return
# Eval.
try:
result = self.eval(text)
except KeyboardInterrupt as e: # KeyboardInterrupt doesn't inherit from Exception.
result = self.eval(expression)
except KeyboardInterrupt: # KeyboardInterrupt doesn't inherit from Exception.
raise
except SystemExit:
return
raise
except BaseException as e:
self._handle_exception(e)
else:
@ -135,11 +118,53 @@ class PythonRepl(PythonInput):
# prevent that a Control-C keypress terminates the REPL in
# any case.)
self._handle_keyboard_interrupt(e)
def run(self) -> None:
"""
Run the REPL loop.
"""
if self.terminal_title:
set_title(self.terminal_title)
self._add_to_namespace()
try:
while True:
# Pull text from the user.
try:
text = self.read()
except EOFError:
return
# Run it; display the result (or errors if applicable).
self.run_and_show_expression(text)
finally:
if self.terminal_title:
clear_title()
self._remove_from_namespace()
async def run_and_show_expression_async(self, text):
loop = asyncio.get_event_loop()
try:
result = await self.eval_async(text)
except KeyboardInterrupt: # KeyboardInterrupt doesn't inherit from Exception.
raise
except SystemExit:
return
except BaseException as e:
self._handle_exception(e)
else:
# Print.
if result is not None:
await loop.run_in_executor(None, lambda: self.show_result(result))
# Loop.
self.current_statement_index += 1
self.signatures = []
# Return the result for future consumers.
return result
async def run_async(self) -> None:
"""
Run the REPL loop, but run the blocking parts in an executor, so that
@ -169,24 +194,7 @@ class PythonRepl(PythonInput):
return
# Eval.
try:
result = await self.eval_async(text)
except KeyboardInterrupt as e: # KeyboardInterrupt doesn't inherit from Exception.
raise
except SystemExit:
return
except BaseException as e:
self._handle_exception(e)
else:
# Print.
if result is not None:
await loop.run_in_executor(
None, lambda: self.show_result(result)
)
# Loop.
self.current_statement_index += 1
self.signatures = []
await self.run_and_show_expression_async(text)
except KeyboardInterrupt as e:
# XXX: This does not yet work properly. In some situations,
@ -231,7 +239,10 @@ class PythonRepl(PythonInput):
# above, then `sys.exc_info()` would not report the right error.
# See issue: https://github.com/prompt-toolkit/ptpython/issues/435
code = self._compile_with_flags(line, "exec")
exec(code, self.get_globals(), self.get_locals())
result = eval(code, self.get_globals(), self.get_locals())
if _has_coroutine_flag(code):
result = asyncio.get_event_loop().run_until_complete(result)
return None
@ -263,9 +274,14 @@ class PythonRepl(PythonInput):
self._store_eval_result(result)
return result
# If not a valid `eval` expression, run using `exec` instead.
# If not a valid `eval` expression, compile as `exec` expression
# but still run with eval to get an awaitable in case of a
# awaitable expression.
code = self._compile_with_flags(line, "exec")
exec(code, self.get_globals(), self.get_locals())
result = eval(code, self.get_globals(), self.get_locals())
if _has_coroutine_flag(code):
result = await result
return None
@ -286,9 +302,9 @@ class PythonRepl(PythonInput):
dont_inherit=True,
)
def show_result(self, result: object) -> None:
def _format_result_output(self, result: object) -> StyleAndTextTuples:
"""
Show __repr__ for an `eval` result.
Format __repr__ for an `eval` result.
Note: this can raise `KeyboardInterrupt` if either calling `__repr__`,
`__pt_repr__` or formatting the output with "Black" takes to long
@ -304,7 +320,7 @@ class PythonRepl(PythonInput):
except BaseException as e:
# Calling repr failed.
self._handle_exception(e)
return
return []
try:
compile(result_repr, "", "eval")
@ -315,11 +331,14 @@ class PythonRepl(PythonInput):
if self.enable_output_formatting:
# Inline import. Slightly speed up start-up time if black is
# not used.
try:
import black
except ImportError:
pass # no Black package in your installation
else:
result_repr = black.format_str(
result_repr,
mode=black.FileMode(line_length=self.app.output.get_size().columns),
mode=black.Mode(line_length=self.app.output.get_size().columns),
)
formatted_result_repr = to_formatted_text(
@ -363,10 +382,18 @@ class PythonRepl(PythonInput):
out_prompt + [("", fragment_list_to_text(formatted_result_repr))]
)
return to_formatted_text(formatted_output)
def show_result(self, result: object) -> None:
"""
Show __repr__ for an `eval` result and print to ouptut.
"""
formatted_text_output = self._format_result_output(result)
if self.enable_pager:
self.print_paginated_formatted_text(to_formatted_text(formatted_output))
self.print_paginated_formatted_text(formatted_text_output)
else:
self.print_formatted_text(to_formatted_text(formatted_output))
self.print_formatted_text(formatted_text_output)
self.app.output.flush()
@ -421,15 +448,7 @@ class PythonRepl(PythonInput):
# Run pager prompt in another thread.
# Same as for the input. This prevents issues with nested event
# loops.
pager_result = None
def in_thread() -> None:
nonlocal pager_result
pager_result = pager_prompt.prompt()
th = threading.Thread(target=in_thread)
th.start()
th.join()
pager_result = pager_prompt.prompt(in_thread=True)
if pager_result == PagerResult.ABORT:
print("...")
@ -494,9 +513,7 @@ class PythonRepl(PythonInput):
"""
return create_pager_prompt(self._current_style, self.title)
def _handle_exception(self, e: BaseException) -> None:
output = self.app.output
def _format_exception_output(self, e: BaseException) -> PygmentsTokens:
# Instead of just calling ``traceback.format_exc``, we take the
# traceback and skip the bottom calls of this framework.
t, v, tb = sys.exc_info()
@ -525,9 +542,15 @@ class PythonRepl(PythonInput):
tokens = list(_lex_python_traceback(tb_str))
else:
tokens = [(Token, tb_str)]
return PygmentsTokens(tokens)
def _handle_exception(self, e: BaseException) -> None:
output = self.app.output
tokens = self._format_exception_output(e)
print_formatted_text(
PygmentsTokens(tokens),
tokens,
style=self._current_style,
style_transformation=self.style_transformation,
include_default_pygments_style=False,
@ -590,7 +613,9 @@ def enable_deprecation_warnings() -> None:
warnings.filterwarnings("default", category=DeprecationWarning, module="__main__")
def run_config(repl: PythonInput, config_file: str = "~/.ptpython/config.py") -> None:
def run_config(
repl: PythonInput, config_file: str = "~/.config/ptpython/config.py"
) -> None:
"""
Execute REPL config file.

12
setup.py
View file

@ -11,7 +11,7 @@ with open(os.path.join(os.path.dirname(__file__), "README.rst")) as f:
setup(
name="ptpython",
author="Jonathan Slenders",
version="3.0.16",
version="3.0.19",
url="https://github.com/prompt-toolkit/ptpython",
description="Python REPL build on top of prompt_toolkit",
long_description=long_description,
@ -20,10 +20,9 @@ setup(
"appdirs",
"importlib_metadata;python_version<'3.8'",
"jedi>=0.16.0",
# Use prompt_toolkit 3.0.16, because of the `DeduplicateCompleter`.
"prompt_toolkit>=3.0.16,<3.1.0",
# Use prompt_toolkit 3.0.18, because of the `in_thread` option.
"prompt_toolkit>=3.0.18,<3.1.0",
"pygments",
"black",
],
python_requires=">=3.6",
classifiers=[
@ -47,5 +46,8 @@ setup(
% sys.version_info[:2],
]
},
extras_require={"ptipython": ["ipython"]}, # For ptipython, we need to have IPython
extras_require={
"ptipython": ["ipython"], # For ptipython, we need to have IPython
"all": ["black"], # Black not always possible on PyPy
},
)