Add a trio repl

docs/source/reference-core.rst

@@ -1903,6 +1903,66 @@ explicit and might be easier to reason about.
  ``contextvars``.
 
 
+.. _interactive debugging:
+
+
+Interactive debugging
+---------------------
+
+When you start an interactive Python session to debug any async program
+(whether it's based on ``asyncio``, Trio, or something else), every await
+expression needs to be inside an async function:
+
+.. code-block:: console
+
+ $ python
+ Python 3.10.6
+ Type "help", "copyright", "credits" or "license" for more information.
+ >>> import trio
+ >>> await trio.sleep(1)
+ File "<stdin>", line 1
+ SyntaxError: 'await' outside function
+ >>> async def main():
+ ... print("hello...")
+ ... await trio.sleep(1)
+ ... print("world!")
+ ...
+ >>> trio.run(main)
+ hello...
+ world!
+
+This can make it difficult to iterate quickly since you have to redefine the
+whole function body whenever you make a tweak.
+
+Trio provides a modified interactive console that lets you ``await`` at the top
+level. You can access this console by running ``python -m trio``:
+
+.. code-block:: console
+
+ $ python -m trio
+ Trio 0.21.0+dev, Python 3.10.6
+ Use "await" directly instead of "trio.run()".
+ Type "help", "copyright", "credits" or "license" for more information.
+ >>> import trio
+ >>> print("hello..."); await trio.sleep(1); print("world!")
+ hello...
+ world!
+
+If you are an IPython user, you can use IPython's `autoawait
+<https://ipython.readthedocs.io/en/stable/interactive/magics.html#magic-autoawait>`__
+function. This can be enabled within the IPython shell by running the magic command
+``%autoawait trio``. To have ``autoawait`` enabled whenever Trio installed, you can
+add the following to your IPython startup files.
+(e.g. ``~/.ipython/profile_default/startup/10-async.py``)
+
+.. code-block::
+
+ try:
+ import trio
+ get_ipython().run_line_magic("autoawait", "trio")
+ except ImportError:
+ pass
+
 Exceptions and warnings
 -----------------------
 

newsfragments/2972.feature.rst

@@ -0,0 +1,16 @@
+Added an interactive interpreter ``python -m trio``.
+
+This makes it easier to try things and experiment with trio in the a Python repl.
+Use the ``await`` keyword without needing to call ``trio.run()``
+
+.. code-block:: console
+
+ $ python -m trio
+ Trio 0.21.0+dev, Python 3.10.6
+ Use "await" directly instead of "trio.run()".
+ Type "help", "copyright", "credits" or "license" for more information.
+ >>> import trio
+ >>> await trio.sleep(1); print("hi") # prints after one second
+ hi
+
+See :ref:`interactive debugging` for further detail.

src/trio/__main__.py

@@ -0,0 +1,3 @@
+from trio._repl import main
+
+main(locals())

src/trio/_repl.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import ast
+import contextlib
+import inspect
+import sys
+import types
+import warnings
+from code import InteractiveConsole
+
+import trio
+import trio.lowlevel
+
+
+class TrioInteractiveConsole(InteractiveConsole):
+ # code.InteractiveInterpreter defines locals as Mapping[str, Any]
+ # but when we pass this to FunctionType it expects a dict. So
+ # we make the type more specific on our subclass
+ locals: dict[str, object]
+
+ def __init__(self, repl_locals: dict[str, object] | None = None):
+ super().__init__(locals=repl_locals)
+ self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
+
+ def runcode(self, code: types.CodeType) -> None:
+ async def _runcode_in_trio() -> BaseException | None:
+ func = types.FunctionType(code, self.locals)
+ try:
+ coro = func()
+ except BaseException as e:
+ return e
+
+ if inspect.iscoroutine(coro):
+ try:
+ await coro
+ except BaseException as e:
+ return e
+ return None
+
+ maybe_exc_or_excgroup = trio.from_thread.run(_runcode_in_trio)
+
+ if maybe_exc_or_excgroup is not None:
+ # maybe_exc_or_excgroup is an exception, or an exception group.
+ # If it is SystemExit quit the repl. Otherwise, print the
+ # traceback.
+ # There could be a SystemExit inside a BaseExceptionGroup. If
+ # that happens, it probably isn't the user trying to quit the
+ # repl, but an error in the code. So we print the exception
+ # and stay in the repl.
+ if isinstance(maybe_exc_or_excgroup, SystemExit):
+ raise maybe_exc_or_excgroup
+
+ # If we didn't raise above, there was an exception, but no
+ # SystemExit. So we raise here and except, so that the console
+ # can print the traceback to the user.
+ try:
+ raise maybe_exc_or_excgroup
+ except BaseException:
+ self.showtraceback()
+
+
+async def run_repl(console: TrioInteractiveConsole) -> None:
+ banner = (
+ f"trio REPL {sys.version} on {sys.platform}\n"
+ f'Use "await" directly instead of "trio.run()".\n'
+ f'Type "help", "copyright", "credits" or "license" '
+ f"for more information.\n"
+ f'{getattr(sys, "ps1", ">>> ")}import trio'
+ )
+ try:
+ await trio.to_thread.run_sync(console.interact, banner)
+ finally:
+ warnings.filterwarnings(
+ "ignore",
+ message=r"^coroutine .* was never awaited$",
+ category=RuntimeWarning,
+ )
+
+
+def main(original_locals: dict[str, object]) -> None:
+ with contextlib.suppress(ImportError):
+ import readline # noqa: F401
+
+ repl_locals: dict[str, object] = {"trio": trio}
+ for key in {
+ "__name__",
+ "__package__",
+ "__loader__",
+ "__spec__",
+ "__builtins__",
+ "__file__",
+ }:
+ repl_locals[key] = original_locals[key]
+
+ console = TrioInteractiveConsole(repl_locals)
+ trio.run(run_repl, console)

src/trio/_tests/test_repl.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+from typing import Protocol
+
+import pytest
+
+import trio._repl
+
+
+class RawInput(Protocol):
+ def __call__(self, prompt: str = "") -> str: ...
+
+
+def build_raw_input(cmds: list[str]) -> RawInput:
+ """
+ Pass in a list of strings.
+ Returns a callable that returns each string, each time its called
+ When there are not more strings to return, raise EOFError
+ """
+ cmds_iter = iter(cmds)
+ prompts = []
+
+ def _raw_helper(prompt: str = "") -> str:
+ prompts.append(prompt)
+ try:
+ return next(cmds_iter)
+ except StopIteration:
+ raise EOFError from None
+
+ return _raw_helper
+
+
+def test_build_raw_input() -> None:
+ """Quick test of our helper function."""
+ raw_input = build_raw_input(["cmd1"])
+ assert raw_input() == "cmd1"
+ with pytest.raises(EOFError):
+ raw_input()
+
+
+# In 3.10 or later, types.FunctionType (used internally) will automatically
+# attach __builtins__ to the function objects. However we need to explicitly
+# include it for 3.8 & 3.9
+def build_locals() -> dict[str, object]:
+ return {"__builtins__": __builtins__}
+
+
+async def test_basic_interaction(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """
+ Run some basic commands through the interpreter while capturing stdout.
+ Ensure that the interpreted prints the expected results.
+ """
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ # evaluate simple expression and recall the value
+ "x = 1",
+ "print(f'{x=}')",
+ # Literal gets printed
+ "'hello'",
+ # define and call sync function
+ "def func():",
+ " print(x + 1)",
+ "",
+ "func()",
+ # define and call async function
+ "async def afunc():",
+ " return 4",
+ "",
+ "await afunc()",
+ # import works
+ "import sys",
+ "sys.stdout.write('hello stdout\\n')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ assert out.splitlines() == ["x=1", "'hello'", "2", "4", "hello stdout", "13"]
+
+
+async def test_system_exits_quit_interpreter(monkeypatch: pytest.MonkeyPatch) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ "raise SystemExit",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ with pytest.raises(SystemExit):
+ await trio._repl.run_repl(console)
+
+
+async def test_system_exits_in_exc_group(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ "import sys",
+ "if sys.version_info < (3, 11):",
+ " from exceptiongroup import BaseExceptionGroup",
+ "",
+ "raise BaseExceptionGroup('', [RuntimeError(), SystemExit()])",
+ "print('AFTER BaseExceptionGroup')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ # assert that raise SystemExit in an exception group
+ # doesn't quit
+ assert "AFTER BaseExceptionGroup" in out
+
+
+async def test_system_exits_in_nested_exc_group(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ "import sys",
+ "if sys.version_info < (3, 11):",
+ " from exceptiongroup import BaseExceptionGroup",
+ "",
+ "raise BaseExceptionGroup(",
+ " '', [BaseExceptionGroup('', [RuntimeError(), SystemExit()])])",
+ "print('AFTER BaseExceptionGroup')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ # assert that raise SystemExit in an exception group
+ # doesn't quit
+ assert "AFTER BaseExceptionGroup" in out
+
+
+async def test_base_exception_captured(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ # The statement after raise should still get executed
+ "raise BaseException",
+ "print('AFTER BaseException')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ assert "AFTER BaseException" in out
+
+
+async def test_exc_group_captured(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ # The statement after raise should still get executed
+ "raise ExceptionGroup('', [KeyError()])",
+ "print('AFTER ExceptionGroup')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ assert "AFTER ExceptionGroup" in out
+
+
+async def test_base_exception_capture_from_coroutine(
+ capsys: pytest.CaptureFixture[str],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ console = trio._repl.TrioInteractiveConsole(repl_locals=build_locals())
+ raw_input = build_raw_input(
+ [
+ "async def async_func_raises_base_exception():",
+ " raise BaseException",
+ "",
+ # This will raise, but the statement after should still
+ # be executed
+ "await async_func_raises_base_exception()",
+ "print('AFTER BaseException')",
+ ]
+ )
+ monkeypatch.setattr(console, "raw_input", raw_input)
+ await trio._repl.run_repl(console)
+ out, err = capsys.readouterr()
+ assert "AFTER BaseException" in out
+
+
+def test_main_entrypoint() -> None:
+ """
+ Basic smoke test when running via the package __main__ entrypoint.
+ """
+ repl = subprocess.run([sys.executable, "-m", "trio"], input=b"exit()")
+ assert repl.returncode == 0