Streams

A “stream” in AnyIO is a simple interface for transporting information from one place to another. It can mean either in-process communication or sending data over a network. AnyIO divides streams into two categories: byte streams and object streams.

Byte streams (“Streams” in Trio lingo) are objects that receive and/or send chunks of bytes. They are modelled after the limitations of the stream sockets, meaning the boundaries are not respected. In practice this means that if, for example, you call .send(b'hello ') and then .send(b'world'), the other end will receive the data chunked in any arbitrary way, like (b'hello' and b' world'), b'hello world' or (b'hel', b'lo wo', b'rld').

Object streams (“Channels” in Trio lingo), on the other hand, deal with Python objects. The most commonly used implementation of these is the memory object stream. The exact semantics of object streams vary a lot by implementation.

Many stream implementations wrap other streams. Of these, some can wrap any bytes-oriented streams, meaning ObjectStream[bytes] and ByteStream. This enables many interesting use cases.

Memory object streams

Memory object streams are intended for implementing a producer-consumer pattern with multiple tasks. Using create_memory_object_stream(), you get a pair of object streams: one for sending, one for receiving. They essentially work like queues, but with support for closing and asynchronous iteration.

By default, memory object streams are created with a buffer size of 0. This means that send() will block until there’s another task that calls receive(). You can set the buffer size to a value of your choosing when creating the stream. It is also possible to have an unbounded buffer by passing math.inf as the buffer size but this is not recommended.

Memory object streams can be cloned by calling the clone() method. Each clone can be closed separately, but each end of the stream is only considered closed once all of its clones have been closed. For example, if you have two clones of the receive stream, the send stream will start raising BrokenResourceError only when both receive streams have been closed.

Multiple tasks can send and receive on the same memory object stream (or its clones) but each sent item is only ever delivered to a single recipient.

The receive ends of memory object streams can be iterated using the async iteration protocol. The loop exits when all clones of the send stream have been closed.

Example:

from anyio import create_task_group, create_memory_object_stream, run
from anyio.streams.memory import MemoryObjectReceiveStream


async def process_items(receive_stream: MemoryObjectReceiveStream[str]) -> None:
    async with receive_stream:
        async for item in receive_stream:
            print('received', item)


async def main():
    # The [str] specifies the type of the objects being passed through the
    # memory object stream. This is a bit of trick, as create_memory_object_stream
    # is actually a class masquerading as a function.
    send_stream, receive_stream = create_memory_object_stream[str]()
    async with create_task_group() as tg:
        tg.start_soon(process_items, receive_stream)
        async with send_stream:
            for num in range(10):
                await send_stream.send(f'number {num}')

run(main)

In contrast to other AnyIO streams (but in line with Trio’s Channels), memory object streams can be closed synchronously, using either the close() method or by using the stream as a context manager:

from anyio.streams.memory import MemoryObjectSendStream


def synchronous_callback(send_stream: MemoryObjectSendStream[str]) -> None:
    with send_stream:
        send_stream.send_nowait('hello')

Stapled streams

A stapled stream combines any mutually compatible receive and send stream together, forming a single bidirectional stream.

It comes in two variants:

StapledByteStream (combines a ByteReceiveStream with a ByteSendStream)
StapledObjectStream (combines an ObjectReceiveStream with a compatible ObjectSendStream)

Buffered byte streams

A buffered byte stream wraps an existing bytes-oriented receive stream and provides certain amenities that require buffering, such as receiving an exact number of bytes, or receiving until the given delimiter is found.

Example:

from anyio import run, create_memory_object_stream
from anyio.streams.buffered import BufferedByteReceiveStream


async def main():
    send, receive = create_memory_object_stream[bytes](4)
    buffered = BufferedByteReceiveStream(receive)
    for part in b'hel', b'lo, ', b'wo', b'rld!':
        await send.send(part)

    result = await buffered.receive_exactly(8)
    print(repr(result))

    result = await buffered.receive_until(b'!', 10)
    print(repr(result))

run(main)

The above script gives the following output:

b'hello, w'
b'orld'

Text streams

Text streams wrap existing receive/send streams and encode/decode strings to bytes and vice versa.

Example:

from anyio import run, create_memory_object_stream
from anyio.streams.text import TextReceiveStream, TextSendStream


async def main():
    bytes_send, bytes_receive = create_memory_object_stream[bytes](1)
    text_send = TextSendStream(bytes_send)
    await text_send.send('åäö')
    result = await bytes_receive.receive()
    print(repr(result))

    text_receive = TextReceiveStream(bytes_receive)
    await bytes_send.send(result)
    result = await text_receive.receive()
    print(repr(result))

run(main)

The above script gives the following output:

b'\xc3\xa5\xc3\xa4\xc3\xb6'
'åäö'

File streams

File streams read from or write to files on the file system. They can be useful for substituting a file for another source of data, or writing output to a file for logging or debugging purposes.

Example:

from anyio import run
from anyio.streams.file import FileReadStream, FileWriteStream


async def main():
    path = '/tmp/testfile'
    async with await FileWriteStream.from_path(path) as stream:
        await stream.send(b'Hello, World!')

    async with await FileReadStream.from_path(path) as stream:
        async for chunk in stream:
            print(chunk.decode(), end='')

    print()

run(main)

New in version 3.0.

TLS streams

TLS (Transport Layer Security), the successor to SSL (Secure Sockets Layer), is the supported way of providing authenticity and confidentiality for TCP streams in AnyIO.

TLS is typically established right after the connection has been made. The handshake involves the following steps:

Sending the certificate to the peer (usually just by the server)
Checking the peer certificate(s) against trusted CA certificates
Checking that the peer host name matches the certificate

Obtaining a server certificate

There are three principal ways you can get an X.509 certificate for your server:

Create a self signed certificate
Use certbot or a similar software to automatically obtain certificates from Let’s Encrypt
Buy one from a certificate vendor

The first option is probably the easiest, but this requires that any client connecting to your server adds the self signed certificate to their list of trusted certificates. This is of course impractical outside of local development and is strongly discouraged in production use.

The second option is nowadays the recommended method, as long as you have an environment where running certbot or similar software can automatically replace the certificate with a newer one when necessary, and that you don’t need any extra features like class 2 validation.

The third option may be your only valid choice when you have special requirements for the certificate that only a certificate vendor can fulfill, or that automatically renewing the certificates is not possible or practical in your environment.

Using self signed certificates

To create a self signed certificate for localhost, you can use the openssl command line tool:

openssl req -x509 -newkey rsa:2048 -subj '/CN=localhost' -keyout key.pem -out cert.pem -nodes -days 365

This creates a (2048 bit) private RSA key (key.pem) and a certificate (cert.pem) matching the host name “localhost”. The certificate will be valid for one year with these settings.

To set up a server using this key-certificate pair:

import ssl

from anyio import create_tcp_listener, run
from anyio.streams.tls import TLSListener


async def handle(client):
    async with client:
        name = await client.receive()
        await client.send(b'Hello, %s\n' % name)


async def main():
    # Create a context for the purpose of authenticating clients
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)

    # Load the server certificate and private key
    context.load_cert_chain(certfile='cert.pem', keyfile='key.pem')

    # Create the listener and start serving connections
    listener = TLSListener(await create_tcp_listener(local_port=1234), context)
    await listener.serve(handle)

run(main)

Connecting to this server can then be done as follows:

import ssl

from anyio import connect_tcp, run


async def main():
    # These two steps are only required for certificates that are not trusted by the
    # installed CA certificates on your machine, so you can skip this part if you
    # use Let's Encrypt or a commercial certificate vendor
    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    context.load_verify_locations(cafile='cert.pem')

    async with await connect_tcp('localhost', 1234, ssl_context=context) as client:
        await client.send(b'Client\n')
        response = await client.receive()
        print(response)

run(main)

Creating self-signed certificates on the fly

When testing your TLS enabled service, it would be convenient to generate the certificates on the fly. To this end, you can use the trustme library:

import ssl

import pytest
import trustme


@pytest.fixture(scope='session')
def ca():
    return trustme.CA()


@pytest.fixture(scope='session')
def server_context(ca):
    server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ca.issue_cert('localhost').configure_cert(server_context)
    return server_context


@pytest.fixture(scope='session')
def client_context(ca):
    client_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    ca.configure_trust(client_context)
    return client_context

You can then pass the server and client contexts from the above fixtures to TLSListener, wrap() or whatever you use on either side.

Dealing with ragged EOFs

According to the TLS standard, encrypted connections should end with a closing handshake. This practice prevents so-called truncation attacks. However, broadly available implementations for protocols such as HTTP, widely ignore this requirement because the protocol level closing signal would make the shutdown handshake redundant.

AnyIO follows the standard by default (unlike the Python standard library’s ssl module). The practical implication of this is that if you’re implementing a protocol that is expected to skip the TLS closing handshake, you need to pass the standard_compatible=False option to wrap() or TLSListener.