Asyncio Streams in Python

December 22, 2022 Python Asyncio

A major benefit of asyncio is the ability to perform non-blocking reads and writes with sockets.

This can be used with network programming applications, such as client programs that read and write with remote servers via APIs, with programs that connect to servers using common protocols, and by developing server applications.

Central to socket programming in asyncio are streams for reading and writing data. A solid knowledge of streams is required in developing client-side asyncio programs, as well as servers.

In this tutorial, you will discover how to use asyncio streams.

After completing this tutorial, you will know:

What are asyncio streams and how to open a TCP socket connection.
How to read and write data using asyncio streams and ensure write buffers are empty.
How to close an open connection and ensure the connection is closed before moving on.

Let's get started.

Asyncio Streams

Asyncio provides non-blocking I/O socket programming.

This is provided via streams.

Streams are high-level async/await-ready primitives to work with network connections. Streams allow sending and receiving data without using callbacks or low-level protocols and transports.
-- Asyncio Streams

Sockets can be opened that provide access to a stream writer and a stream writer.

Data can then be written and read from the stream using coroutines, suspending when appropriate.

Once finished, the socket can be closed.

The asyncio streams capability is low-level meaning that any protocols required must be implemented manually.

This might include common web protocols, such as:

HTTP or HTTPS for interacting with web servers
SMTP for interacting with email servers
FTP for interacting with file servers.

The streams can also be used to create a server to handle requests using a standard protocol, or to develop your own application-specific protocol.

Now that we know what asyncio streams are, let's look at how to use them.

How to Open a Connection

An asyncio TCP client socket connection can be opened using the asyncio.open_connection() function.

Establish a network connection and return a pair of (reader, writer) objects. The returned reader and writer objects are instances of StreamReader and StreamWriter classes.
-- Asyncio Streams

This is a coroutine that must be awaited and will return once the socket connection is open.

The function returns a StreamReader and StreamWriter object for interacting with the socket.

For example:

...
# open a connection
reader, writer = await asyncio.open_connection(...)

The asyncio.open_connection() function takes many arguments in order to configure the socket connection.

The two required arguments are the host and the port.

The host is a string that specifies the server to connect to, such as a domain name or an IP address.

The port is the socket port number, such as 80 for HTTP servers, 443 for HTTPS servers, 23 for SMTP and so on.

For example:

...
# open a connection to an http server
reader, writer = await asyncio.open_connection('www.google.com', 80)

Encrypted socket connections are supported over the SSL protocol.

The most common example might be HTTPS which is replacing HTTP.

This can be achieved by setting the "ssl" argument to True.

For example:

...
# open a connection to an https server
reader, writer = await asyncio.open_connection('www.google.com', 443, ssl=True)

How to Start a Server

An asyncio TCP server socket can be opened using the asyncio.start_server() function.

Create a TCP server (socket type SOCK_STREAM) listening on port of the host address.
-- Asyncio Event Loop

This is a coroutine that must be awaited.

The function returns an asyncio.Server object that represents the running server.

For example:

...
# start a tcp server
server = await asyncio.start_server(...)

The three required arguments are the callback function, the host, and the port.

The callback function is a custom function specified by name that will be called each time a client connects to the server.

The client_connected_cb callback is called whenever a new client connection is established. It receives a (reader, writer) pair as two arguments, instances of the StreamReader and StreamWriter classes.
-- Asyncio Streams

The host is the domain name or IP address that clients will specify to connect. The port is the socket port number on which to receive connections, such as 21 for FTP or 80 for HTTP.

For example:

# handle connections
async def handler(reader, writer):
	# ...

...
# start a server to receive http connections
server = await asyncio.start_server(handler, '127.0.0.1', 80)

How to Write Data with the StreamWriter

We can write data to the socket using an asyncio.StreamWriter.

Represents a writer object that provides APIs to write data to the IO stream.
-- Asyncio Streams

Data is written as bytes.

Byte data can be written to the socket using the write() method.

The method attempts to write the data to the underlying socket immediately. If that fails, the data is queued in an internal write buffer until it can be sent.
-- Asyncio Streams

For example:

...
# write byte data
writer.write(byte_data)

Alternatively, multiple "lines" of byte data organized into a list or iterable can be written using the writelines() method.

For example:

...
# write lines of byte data
writer.writelines(byte_lines)

Neither method for writing data blocks or suspends the calling coroutine.

After writing byte data it is a good idea to drain the socket via the drain() method.

Wait until it is appropriate to resume writing to the stream.
-- Asyncio Streams

This is a coroutine and will suspend the caller until the bytes have been transmitted and the socket is ready.

For example:

...
# write byte data
writer.write(byte_data)
# wait for data to be transmitted
await writer.drain()

How to Read Data with the StreamReader

We can read data from the socket using an asyncio.StreamReader.

Represents a reader object that provides APIs to read data from the IO stream.
-- Asyncio Streams

Data is read in byte format, therefore strings may need to be encoded before being used.

All read methods are coroutines that must be awaited.

An arbitrary number of bytes can be read via the read() method, which will read until the end of file (EOF).

...
# read byte data
byte_data = await reader.read()

Additionally, the number of bytes to read can be specified via the "n" argument.

Read up to n bytes. If n is not provided, or set to -1, read until EOF and return all read bytes.
-- Asyncio Streams

This may be helpful if you know the number of bytes expected from the next response.

For example:

...
# read byte data
byte_data = await reader.read(n=100)

A single line of data can be read using the readline() method.

This will return bytes until a new line character '\n' is encountered, or EOF.

Read one line, where “line” is a sequence of bytes ending with \n. If EOF is received and \n was not found, the method returns partially read data. If EOF is received and the internal buffer is empty, return an empty bytes object.
-- Asyncio Streams

This is helpful when reading standard protocols that operate with lines of text.

...
# read a line data
byte_line = await reader.readline()

Additionally, there is a readexactly() method to read an exact number of bytes otherwise raise an exception, and a readuntil() that will read bytes until a specified character in byte form is read.

How to Close Connection

The socket can be closed via the asyncio.StreamWriter.

The close() method can be called which will close the socket.

The method closes the stream and the underlying socket.
-- Asyncio Streams

This method does not block.

For example:

...
# close the socket
writer.close()

Although the close() method does not block, we can wait for the socket to close completely before continuing on.

This can be achieved via the wait_closed() method.

Wait until the stream is closed. Should be called after close() to wait until the underlying connection is closed.
-- Asyncio Streams

This is a coroutine that can be awaited.

For example:

...
# close the socket
writer.close()
# wait for the socket to close
await writer.wait_closed()

We can check if the socket has been closed or is in the process of being closed via the is_closing() method.

For example:

...
# check if the socket is closed or closing
if writer.is_closing():
	# ...

Takeaways

You now know how to use asyncio streams in Python.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.