Streams

Source code:Lib/asyncio/streams.py


Streams are high-level async/await-ready primitives to work withnetwork connections. Streams allow sending and receiving data withoutusing callbacks or low-level protocols and transports.

Here is an example of a TCP echo client written using asynciostreams:

importasyncioasyncdeftcp_echo_client(message):reader,writer=awaitasyncio.open_connection('127.0.0.1',8888)print(f'Send:{message!r}')writer.write(message.encode())awaitwriter.drain()data=awaitreader.read(100)print(f'Received:{data.decode()!r}')print('Close the connection')writer.close()awaitwriter.wait_closed()asyncio.run(tcp_echo_client('Hello World!'))

See also theExamples section below.

Stream Functions

The following top-level asyncio functions can be used to createand work with streams:

coroutineasyncio.open_connection(host=None,port=None,*,loop=None,limit=None,ssl=None,family=0,proto=0,flags=0,sock=None,local_addr=None,server_hostname=None,ssl_handshake_timeout=None)

Establish a network connection and return a pair of(reader,writer) objects.

The returnedreader andwriter objects are instances ofStreamReader andStreamWriter classes.

Theloop argument is optional and can always be determinedautomatically when this function is awaited from a coroutine.

limit determines the buffer size limit used by thereturnedStreamReader instance. By default thelimitis set to 64 KiB.

The rest of the arguments are passed directly toloop.create_connection().

New in version 3.7:Thessl_handshake_timeout parameter.

coroutineasyncio.start_server(client_connected_cb,host=None,port=None,*,loop=None,limit=None,family=socket.AF_UNSPEC,flags=socket.AI_PASSIVE,sock=None,backlog=100,ssl=None,reuse_address=None,reuse_port=None,ssl_handshake_timeout=None,start_serving=True)

Start a socket server.

Theclient_connected_cb callback is called whenever a new clientconnection is established. It receives a(reader,writer) pairas two arguments, instances of theStreamReader andStreamWriter classes.

client_connected_cb can be a plain callable or acoroutine function; if it is a coroutine function,it will be automatically scheduled as aTask.

Theloop argument is optional and can always be determinedautomatically when this method is awaited from a coroutine.

limit determines the buffer size limit used by thereturnedStreamReader instance. By default thelimitis set to 64 KiB.

The rest of the arguments are passed directly toloop.create_server().

New in version 3.7:Thessl_handshake_timeout andstart_serving parameters.

Unix Sockets

coroutineasyncio.open_unix_connection(path=None,*,loop=None,limit=None,ssl=None,sock=None,server_hostname=None,ssl_handshake_timeout=None)

Establish a Unix socket connection and return a pair of(reader,writer).

Similar toopen_connection() but operates on Unix sockets.

See also the documentation ofloop.create_unix_connection().

Availability: Unix.

New in version 3.7:Thessl_handshake_timeout parameter.

Changed in version 3.7:Thepath parameter can now be apath-like object

coroutineasyncio.start_unix_server(client_connected_cb,path=None,*,loop=None,limit=None,sock=None,backlog=100,ssl=None,ssl_handshake_timeout=None,start_serving=True)

Start a Unix socket server.

Similar tostart_server() but works with Unix sockets.

See also the documentation ofloop.create_unix_server().

Availability: Unix.

New in version 3.7:Thessl_handshake_timeout andstart_serving parameters.

Changed in version 3.7:Thepath parameter can now be apath-like object.

StreamReader

classasyncio.StreamReader

Represents a reader object that provides APIs to read datafrom the IO stream.

It is not recommended to instantiateStreamReader objectsdirectly; useopen_connection() andstart_server()instead.

coroutineread(n=-1)

Read up ton bytes. Ifn is not provided, or set to-1,read until EOF and return all read bytes.

If EOF was received and the internal buffer is empty,return an emptybytes object.

coroutinereadline()

Read one line, where “line” is a sequence of bytesending with\n.

If EOF is received and\n was not found, the methodreturns partially read data.

If EOF is received and the internal buffer is empty,return an emptybytes object.

coroutinereadexactly(n)

Read exactlyn bytes.

Raise anIncompleteReadError if EOF is reached beforencan be read. Use theIncompleteReadError.partialattribute to get the partially read data.

coroutinereaduntil(separator=b'\n')

Read data from the stream untilseparator is found.

On success, the data and separator will be removed from theinternal buffer (consumed). Returned data will include theseparator at the end.

If the amount of data read exceeds the configured stream limit, aLimitOverrunError exception is raised, and the datais left in the internal buffer and can be read again.

If EOF is reached before the complete separator is found,anIncompleteReadError exception is raised, and the internalbuffer is reset. TheIncompleteReadError.partial attributemay contain a portion of the separator.

New in version 3.5.2.

at_eof()

ReturnTrue if the buffer is empty andfeed_eof()was called.

StreamWriter

classasyncio.StreamWriter

Represents a writer object that provides APIs to write datato the IO stream.

It is not recommended to instantiateStreamWriter objectsdirectly; useopen_connection() andstart_server()instead.

write(data)

The method attempts to write thedata to the underlying socket immediately.If that fails, the data is queued in an internal write buffer until it can besent.

The method should be used along with thedrain() method:

stream.write(data)awaitstream.drain()
writelines(data)

The method writes a list (or any iterable) of bytes to the underlying socketimmediately.If that fails, the data is queued in an internal write buffer until it can besent.

The method should be used along with thedrain() method:

stream.writelines(lines)awaitstream.drain()
close()

The method closes the stream and the underlying socket.

The method should be used along with thewait_closed() method:

stream.close()awaitstream.wait_closed()
can_write_eof()

ReturnTrue if the underlying transport supportsthewrite_eof() method,False otherwise.

write_eof()

Close the write end of the stream after the buffered writedata is flushed.

transport

Return the underlying asyncio transport.

get_extra_info(name,default=None)

Access optional transport information; seeBaseTransport.get_extra_info() for details.

coroutinedrain()

Wait until it is appropriate to resume writing to the stream.Example:

writer.write(data)awaitwriter.drain()

This is a flow control method that interacts with the underlyingIO write buffer. When the size of the buffer reachesthe high watermark,drain() blocks until the size of thebuffer is drained down to the low watermark and writing canbe resumed. When there is nothing to wait for, thedrain()returns immediately.

is_closing()

ReturnTrue if the stream is closed or in the process ofbeing closed.

New in version 3.7.

coroutinewait_closed()

Wait until the stream is closed.

Should be called afterclose() to wait until the underlyingconnection is closed.

New in version 3.7.

Examples

TCP echo client using streams

TCP echo client using theasyncio.open_connection() function:

importasyncioasyncdeftcp_echo_client(message):reader,writer=awaitasyncio.open_connection('127.0.0.1',8888)print(f'Send:{message!r}')writer.write(message.encode())data=awaitreader.read(100)print(f'Received:{data.decode()!r}')print('Close the connection')writer.close()asyncio.run(tcp_echo_client('Hello World!'))

See also

TheTCP echo client protocolexample uses the low-levelloop.create_connection() method.

TCP echo server using streams

TCP echo server using theasyncio.start_server() function:

importasyncioasyncdefhandle_echo(reader,writer):data=awaitreader.read(100)message=data.decode()addr=writer.get_extra_info('peername')print(f"Received{message!r} from{addr!r}")print(f"Send:{message!r}")writer.write(data)awaitwriter.drain()print("Close the connection")writer.close()asyncdefmain():server=awaitasyncio.start_server(handle_echo,'127.0.0.1',8888)addr=server.sockets[0].getsockname()print(f'Serving on{addr}')asyncwithserver:awaitserver.serve_forever()asyncio.run(main())

See also

TheTCP echo server protocolexample uses theloop.create_server() method.

Get HTTP headers

Simple example querying HTTP headers of the URL passed on the command line:

importasyncioimporturllib.parseimportsysasyncdefprint_http_headers(url):url=urllib.parse.urlsplit(url)ifurl.scheme=='https':reader,writer=awaitasyncio.open_connection(url.hostname,443,ssl=True)else:reader,writer=awaitasyncio.open_connection(url.hostname,80)query=(f"HEAD{url.pathor'/'} HTTP/1.0\r\n"f"Host:{url.hostname}\r\n"f"\r\n")writer.write(query.encode('latin-1'))whileTrue:line=awaitreader.readline()ifnotline:breakline=line.decode('latin1').rstrip()ifline:print(f'HTTP header>{line}')# Ignore the body, close the socketwriter.close()url=sys.argv[1]asyncio.run(print_http_headers(url))

Usage:

pythonexample.pyhttp://example.com/path/page.html

or with HTTPS:

pythonexample.pyhttps://example.com/path/page.html

Register an open socket to wait for data using streams

Coroutine waiting until a socket receives data using theopen_connection() function:

importasyncioimportsocketasyncdefwait_for_data():# Get a reference to the current event loop because# we want to access low-level APIs.loop=asyncio.get_running_loop()# Create a pair of connected sockets.rsock,wsock=socket.socketpair()# Register the open socket to wait for data.reader,writer=awaitasyncio.open_connection(sock=rsock)# Simulate the reception of data from the networkloop.call_soon(wsock.send,'abc'.encode())# Wait for datadata=awaitreader.read(100)# Got data, we are done: close the socketprint("Received:",data.decode())writer.close()# Close the second socketwsock.close()asyncio.run(wait_for_data())

See also

Theregister an open socket to wait for data using a protocol example uses a low-level protocol andtheloop.create_connection() method.

Thewatch a file descriptor for read events example uses the low-levelloop.add_reader() method to watch a file descriptor.