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:
- coroutine
asyncio.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 of
StreamReaderandStreamWriterclasses.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 thereturned
StreamReaderinstance. By default thelimitis set to 64 KiB.The rest of the arguments are passed directly to
loop.create_connection().New in version 3.7:Thessl_handshake_timeout parameter.
- coroutine
asyncio.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 theStreamReaderandStreamWriterclasses.client_connected_cb can be a plain callable or acoroutine function; if it is a coroutine function,it will be automatically scheduled as a
Task.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 thereturned
StreamReaderinstance. By default thelimitis set to 64 KiB.The rest of the arguments are passed directly to
loop.create_server().New in version 3.7:Thessl_handshake_timeout andstart_serving parameters.
Unix Sockets
- coroutine
asyncio.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 to
open_connection()but operates on Unix sockets.See also the documentation of
loop.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
- coroutine
asyncio.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 to
start_server()but works with Unix sockets.See also the documentation of
loop.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¶
- class
asyncio.StreamReader¶ Represents a reader object that provides APIs to read datafrom the IO stream.
It is not recommended to instantiateStreamReader objectsdirectly; use
open_connection()andstart_server()instead.- coroutine
read(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 empty
bytesobject.
- coroutine
readline()¶ Read one line, where “line” is a sequence of bytesending with
\n.If EOF is received and
\nwas not found, the methodreturns partially read data.If EOF is received and the internal buffer is empty,return an empty
bytesobject.
- coroutine
readexactly(n)¶ Read exactlyn bytes.
Raise an
IncompleteReadErrorif EOF is reached beforencan be read. Use theIncompleteReadError.partialattribute to get the partially read data.
- coroutine
readuntil(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, a
LimitOverrunErrorexception 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,an
IncompleteReadErrorexception is raised, and the internalbuffer is reset. TheIncompleteReadError.partialattributemay contain a portion of the separator.New in version 3.5.2.
at_eof()¶Return
Trueif the buffer is empty andfeed_eof()was called.
- coroutine
StreamWriter¶
- class
asyncio.StreamWriter¶ Represents a writer object that provides APIs to write datato the IO stream.
It is not recommended to instantiateStreamWriter objectsdirectly; use
open_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 the
drain()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 the
drain()method:stream.writelines(lines)awaitstream.drain()
close()¶The method closes the stream and the underlying socket.
The method should be used along with the
wait_closed()method:stream.close()awaitstream.wait_closed()
can_write_eof()¶Return
Trueif the underlying transport supportsthewrite_eof()method,Falseotherwise.
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; see
BaseTransport.get_extra_info()for details.
- coroutine
drain()¶ 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, the
drain()returns immediately.
is_closing()¶Return
Trueif the stream is closed or in the process ofbeing 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.