mirror of https://github.com/scrapy/scrapy.git
145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import code
|
|
from collections.abc import Callable
|
|
from functools import partial, wraps
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterable
|
|
|
|
EmbedFuncT = Callable[..., None]
|
|
KnownShellsT = dict[str, Callable[..., EmbedFuncT]]
|
|
|
|
|
|
def _embed_ipython_shell(
|
|
namespace: dict[str, Any] | None = None, banner: str = ""
|
|
) -> EmbedFuncT:
|
|
"""Start an IPython Shell"""
|
|
from IPython.terminal.embed import InteractiveShellEmbed # noqa: T100,PLC0415
|
|
from IPython.terminal.ipapp import load_default_config # noqa: PLC0415
|
|
|
|
@wraps(_embed_ipython_shell)
|
|
def wrapper(namespace: dict[str, Any] = namespace or {}, banner: str = "") -> None:
|
|
config = load_default_config() # type: ignore[no-untyped-call]
|
|
# Always use .instance() to ensure _instance propagation to all parents
|
|
# this is needed for <TAB> completion works well for new imports
|
|
# and clear the instance to always have the fresh env
|
|
# on repeated breaks like with inspect_response()
|
|
InteractiveShellEmbed.clear_instance()
|
|
shell = InteractiveShellEmbed.instance(
|
|
banner1=banner, user_ns=namespace, config=config
|
|
)
|
|
# If an asyncio event loop is already running in this thread, e.g. when
|
|
# inspect_response() is called from a spider callback while using the
|
|
# asyncio reactor, prompt_toolkit cannot run its own event loop here, so
|
|
# ask it to run the prompt in a separate thread instead. pt_app is None
|
|
# when IPython falls back to its simple prompt, which needs no event loop.
|
|
# See https://github.com/scrapy/scrapy/issues/5447
|
|
if (pt_app := getattr(shell, "pt_app", None)) is not None:
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
pass
|
|
else:
|
|
pt_app.prompt = partial(pt_app.prompt, in_thread=True)
|
|
shell()
|
|
|
|
return wrapper
|
|
|
|
|
|
def _embed_bpython_shell(
|
|
namespace: dict[str, Any] | None = None, banner: str = ""
|
|
) -> EmbedFuncT:
|
|
"""Start a bpython shell"""
|
|
import bpython # noqa: PLC0415
|
|
|
|
@wraps(_embed_bpython_shell)
|
|
def wrapper(namespace: dict[str, Any] = namespace or {}, banner: str = "") -> None:
|
|
bpython.embed(locals_=namespace, banner=banner)
|
|
|
|
return wrapper
|
|
|
|
|
|
def _embed_ptpython_shell(
|
|
namespace: dict[str, Any] | None = None, banner: str = ""
|
|
) -> EmbedFuncT:
|
|
"""Start a ptpython shell"""
|
|
import ptpython.repl # noqa: PLC0415
|
|
|
|
@wraps(_embed_ptpython_shell)
|
|
def wrapper(namespace: dict[str, Any] = namespace or {}, banner: str = "") -> None:
|
|
print(banner)
|
|
ptpython.repl.embed(locals=namespace)
|
|
|
|
return wrapper
|
|
|
|
|
|
def _embed_standard_shell(
|
|
namespace: dict[str, Any] | None = None, banner: str = ""
|
|
) -> EmbedFuncT:
|
|
"""Start a standard python shell"""
|
|
try: # readline module is only available on unix systems
|
|
import readline # noqa: PLC0415
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
import rlcompleter # noqa: F401,PLC0415
|
|
|
|
readline.parse_and_bind("tab:complete") # type: ignore[attr-defined,unused-ignore]
|
|
|
|
@wraps(_embed_standard_shell)
|
|
def wrapper(namespace: dict[str, Any] = namespace or {}, banner: str = "") -> None:
|
|
code.interact(banner=banner, local=namespace)
|
|
|
|
return wrapper
|
|
|
|
|
|
DEFAULT_PYTHON_SHELLS: KnownShellsT = {
|
|
"ptpython": _embed_ptpython_shell,
|
|
"ipython": _embed_ipython_shell,
|
|
"bpython": _embed_bpython_shell,
|
|
"python": _embed_standard_shell,
|
|
}
|
|
|
|
|
|
def get_shell_embed_func(
|
|
shells: Iterable[str] | None = None, known_shells: KnownShellsT | None = None
|
|
) -> EmbedFuncT | None:
|
|
"""Return the first acceptable shell-embed function
|
|
from a given list of shell names.
|
|
"""
|
|
if shells is None: # list, preference order of shells
|
|
shells = DEFAULT_PYTHON_SHELLS.keys()
|
|
if known_shells is None: # available embeddable shells
|
|
known_shells = DEFAULT_PYTHON_SHELLS.copy()
|
|
for shell in shells:
|
|
if shell in known_shells:
|
|
try:
|
|
# function test: run all setup code (imports),
|
|
# but dont fall into the shell
|
|
return known_shells[shell]()
|
|
except ImportError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def start_python_console(
|
|
namespace: dict[str, Any] | None = None,
|
|
banner: str = "",
|
|
shells: Iterable[str] | None = None,
|
|
) -> None:
|
|
"""Start Python console bound to the given namespace.
|
|
Readline support and tab completion will be used on Unix, if available.
|
|
"""
|
|
if namespace is None:
|
|
namespace = {}
|
|
|
|
try:
|
|
shell = get_shell_embed_func(shells)
|
|
if shell is not None:
|
|
shell(namespace=namespace, banner=banner)
|
|
except SystemExit: # raised when using exit() in python code.interact
|
|
pass
|