Subprocess support: high-level interface

newsfragments/833.feature.rst

@@ -0,0 +1,2 @@
+High-level subprocess support: :func:`trio.open_process`, :func:`trio.run_process`,
+:func:`trio.delegate_to_process`.

trio/__init__.py

@@ -36,7 +36,7 @@
     BlockingTrioPortal
 )
 
-from ._highlevel_generic import aclose_forcefully, StapledStream
+from ._highlevel_generic import aclose_forcefully, StapledStream, NullStream
 
 from ._channel import open_memory_channel
 
@@ -64,6 +64,15 @@
     open_ssl_over_tcp_stream, open_ssl_over_tcp_listeners, serve_ssl_over_tcp
 )
 
+from ._subprocess import (
+    Process,
+    ProcessStream,
+    CompletedProcess,
+    open_process,
+    run_process,
+    delegate_to_process,
+)
+
 from ._deprecate import TrioDeprecationWarning
 
 # Imported by default

trio/_deprecated_subprocess_reexports.py

@@ -1,5 +1,3 @@
-from ._subprocess import Process
-
 # Reexport constants and exceptions from the stdlib subprocess module
 from subprocess import (
     PIPE, STDOUT, DEVNULL, CalledProcessError, SubprocessError, TimeoutExpired,

trio/_highlevel_generic.py

@@ -110,3 +110,62 @@ async def aclose(self):
             await self.send_stream.aclose()
         finally:
             await self.receive_stream.aclose()
+
+
+class NullStream(HalfCloseableStream):
+    """A :class:`~trio.abc.HalfCloseableStream` in which any data sent is
+    immediately discarded and receives always return end-of-file.
+
+    This is useful for similar reasons as ``/dev/null`` on Unix; for example,
+    in a function defined to return a stream of data, it reduces special-casing
+    if there's no data that needs to be provided.
+
+    Trying to read after closing, or write after closing or
+    :meth:`~trio.abc.HalfCloseableStream.send_eof`, will raise
+    :exc:`ClosedResourceError`, even though those operations are
+    otherwise no-ops.
+
+    A synchronous ``close()`` is provided in addition to the usual ``aclose()``.
+    """
+
+    def __init__(self) -> None:
+        self._read_closed = False
+        self._write_closed = False
+
+    def __repr__(self) -> str:
+        if self._read_closed:
+            closed = "closed"
+        elif self._write_closed:
+            closed = "sent EOF"
+        else:
+            closed = "open"
+        return "<trio.NullStream: {}>".format(closed)
+
+    def close(self) -> None:
+        self._read_closed = self._write_closed = True
+
+    async def aclose(self) -> None:
+        self.close()
+        await _core.checkpoint()
+
+    async def receive_some(self, max_bytes: int) -> bytes:
+        await _core.checkpoint()
+        if self._read_closed:
+            raise _core.ClosedResourceError
+        return b""
+
+    async def send_all(self, data: bytes) -> None:
+        await _core.checkpoint()
+        if self._write_closed:
+            raise _core.ClosedResourceError
+
+    async def wait_send_all_might_not_block(self) -> None:
+        await _core.checkpoint()
+        if self._write_closed:
+            raise _core.ClosedResourceError
+
+    async def send_eof(self) -> None:
+        await _core.checkpoint()
+        if self._read_closed:
+            raise _core.ClosedResourceError
+        self._write_closed = True