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,114 @@
+from __future__ import annotations
+
+import ast
+import contextlib
+import inspect
+import sys
+import types
+import warnings
+from code import InteractiveConsole
+from typing import Generator
+
+import trio
+import trio.lowlevel
+
+if sys.version_info < (3, 11):
+    from exceptiongroup import BaseExceptionGroup
+
+
+def _flatten_exception_group(
+    excgroup: BaseExceptionGroup[BaseException],
+) -> Generator[BaseException, None, None]:
+    for exc in excgroup.exceptions:
+        if isinstance(exc, BaseExceptionGroup):
+            yield from _flatten_exception_group(exc)
+        else:
+            yield exc
+
+
+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 or if the exception group contains
+            # a SystemExit, quit the repl. Otherwise, print the traceback.
+            if isinstance(maybe_exc_or_excgroup, SystemExit):
+                raise maybe_exc_or_excgroup
+            elif isinstance(maybe_exc_or_excgroup, BaseExceptionGroup):
+                sys_exit_exc = maybe_exc_or_excgroup.subgroup(SystemExit)
+                if sys_exit_exc:
+                    # There is a SystemExit exception, but it might be nested
+                    # If there are more than one SystemExit exception in
+                    # the group, this will only find and re-raise the first.
+                    raise next(_flatten_exception_group(sys_exit_exc))
+
+            # If we didn't raise in either of the conditions 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)