Transports Hierarchy¶

classasyncio.BaseTransport¶

Base class for all transports. Contains methods that allasyncio transports share.

classasyncio.WriteTransport(BaseTransport)¶

A base transport for write-only connections.

Instances of theWriteTransport class are returned fromtheloop.connect_write_pipe() event loop method andare also used by subprocess-related methods likeloop.subprocess_exec().

classasyncio.ReadTransport(BaseTransport)¶

A base transport for read-only connections.

Instances of theReadTransport class are returned fromtheloop.connect_read_pipe() event loop method andare also used by subprocess-related methods likeloop.subprocess_exec().

classasyncio.Transport(WriteTransport,ReadTransport)¶

Interface representing a bidirectional transport, such as aTCP connection.

The user does not instantiate a transport directly; they call autility function, passing it a protocol factory and otherinformation necessary to create the transport and protocol.

Instances of theTransport class are returned from or used byevent loop methods likeloop.create_connection(),loop.create_unix_connection(),loop.create_server(),loop.sendfile(), etc.

classasyncio.DatagramTransport(BaseTransport)¶

A transport for datagram (UDP) connections.

Instances of theDatagramTransport class are returned fromtheloop.create_datagram_endpoint() event loop method.

classasyncio.SubprocessTransport(BaseTransport)¶

An abstraction to represent a connection between a parent and itschild OS process.

Instances of theSubprocessTransport class are returned fromevent loop methodsloop.subprocess_shell() andloop.subprocess_exec().

Base Transport¶

BaseTransport.close()¶

Close the transport.

If the transport has a buffer for outgoingdata, buffered data will be flushed asynchronously. No more datawill be received. After all buffered data is flushed, theprotocol’sprotocol.connection_lost() method will be called withNone as its argument. The transport should not beused once it is closed.

BaseTransport.is_closing()¶

ReturnTrue if the transport is closing or is closed.

BaseTransport.get_extra_info(name,default=None)¶

Return information about the transport or underlying resourcesit uses.

name is a string representing the piece of transport-specificinformation to get.

default is the value to return if the information is notavailable, or if the transport does not support querying itwith the given third-party event loop implementation or on thecurrent platform.

For example, the following code attempts to get the underlyingsocket object of the transport:

sock=transport.get_extra_info('socket')ifsockisnotNone:print(sock.getsockopt(...))

Categories of information that can be queried on some transports:

• socket:

  • 'peername': the remote address to which the socket isconnected, result ofsocket.socket.getpeername()(None on error)

  • 'socket':socket.socket instance

  • 'sockname': the socket’s own address,result ofsocket.socket.getsockname()

• SSL socket:

  • 'compression': the compression algorithm being used as astring, orNone if the connection isn’t compressed; resultofssl.SSLSocket.compression()

  • 'cipher': a three-value tuple containing the name of thecipher being used, the version of the SSL protocol that definesits use, and the number of secret bits being used; result ofssl.SSLSocket.cipher()

  • 'peercert': peer certificate; result ofssl.SSLSocket.getpeercert()

  • 'sslcontext':ssl.SSLContext instance

  • 'ssl_object':ssl.SSLObject orssl.SSLSocket instance

• pipe:

  • 'pipe': pipe object

• subprocess:

  • 'subprocess':subprocess.Popen instance

BaseTransport.set_protocol(protocol)¶

Set a new protocol.

Switching protocol should only be done when bothprotocols are documented to support the switch.

BaseTransport.get_protocol()¶

Return the current protocol.

Read-only Transports¶

ReadTransport.is_reading()¶

ReturnTrue if the transport is receiving new data.

Added in version 3.7.

ReadTransport.pause_reading()¶

Pause the receiving end of the transport. No data will be passed tothe protocol’sprotocol.data_received()method untilresume_reading() is called.

Changed in version 3.7:The method is idempotent, i.e. it can be called when thetransport is already paused or closed.

ReadTransport.resume_reading()¶

Resume the receiving end. The protocol’sprotocol.data_received() methodwill be called once again if some data is available for reading.

Changed in version 3.7:The method is idempotent, i.e. it can be called when thetransport is already reading.

Write-only Transports¶

WriteTransport.abort()¶

Close the transport immediately, without waiting for pending operationsto complete. Buffered data will be lost. No more data will be received.The protocol’sprotocol.connection_lost() method will eventually becalled withNone as its argument.

WriteTransport.can_write_eof()¶

ReturnTrue if the transport supportswrite_eof(),False if not.

WriteTransport.get_write_buffer_size()¶

Return the current size of the output buffer used by the transport.

WriteTransport.get_write_buffer_limits()¶

Get thehigh andlow watermarks for write flow control. Return atuple(low,high) wherelow andhigh are positive number ofbytes.

Useset_write_buffer_limits() to set the limits.

Added in version 3.4.2.

WriteTransport.set_write_buffer_limits(high=None,low=None)¶

Set thehigh andlow watermarks for write flow control.

These two values (measured in number ofbytes) control when the protocol’sprotocol.pause_writing()andprotocol.resume_writing()methods are called. If specified, the low watermark must be lessthan or equal to the high watermark. Neitherhigh norlowcan be negative.

pause_writing() is called when the buffer sizebecomes greater than or equal to thehigh value. If writing hasbeen paused,resume_writing() is called whenthe buffer size becomes less than or equal to thelow value.

The defaults are implementation-specific. If only thehigh watermark is given, the low watermark defaults to animplementation-specific value less than or equal to thehigh watermark. Settinghigh to zero forceslow to zero aswell, and causespause_writing() to be calledwhenever the buffer becomes non-empty. Settinglow to zero causesresume_writing() to be called only once thebuffer is empty. Use of zero for either limit is generallysub-optimal as it reduces opportunities for doing I/O andcomputation concurrently.

Useget_write_buffer_limits()to get the limits.

WriteTransport.write(data)¶

Write somedata bytes to the transport.

This method does not block; it buffers the data and arranges for itto be sent out asynchronously.

WriteTransport.writelines(list_of_data)¶

Write a list (or any iterable) of data bytes to the transport.This is functionally equivalent to callingwrite() on eachelement yielded by the iterable, but may be implemented moreefficiently.

WriteTransport.write_eof()¶

Close the write end of the transport after flushing all buffered data.Data may still be received.

This method can raiseNotImplementedError if the transport(e.g. SSL) doesn’t support half-closed connections.

Datagram Transports¶

DatagramTransport.sendto(data,addr=None)¶

Send thedata bytes to the remote peer given byaddr (atransport-dependent target address). Ifaddr isNone,the data is sent to the target address given on transportcreation.

This method does not block; it buffers the data and arrangesfor it to be sent out asynchronously.

Changed in version 3.13:This method can be called with an empty bytes object to send azero-length datagram. The buffer size calculation used for flowcontrol is also updated to account for the datagram header.

DatagramTransport.abort()¶

Close the transport immediately, without waiting for pendingoperations to complete. Buffered data will be lost.No more data will be received. The protocol’sprotocol.connection_lost()method will eventually be called withNone as its argument.

Subprocess Transports¶

SubprocessTransport.get_pid()¶

Return the subprocess process id as an integer.

SubprocessTransport.get_pipe_transport(fd)¶

Return the transport for the communication pipe corresponding to theinteger file descriptorfd:

• 0: readable streaming transport of the standard input (stdin),orNone if the subprocess was not created withstdin=PIPE

• 1: writable streaming transport of the standard output (stdout),orNone if the subprocess was not created withstdout=PIPE

• 2: writable streaming transport of the standard error (stderr),orNone if the subprocess was not created withstderr=PIPE

• otherfd:None

SubprocessTransport.get_returncode()¶

Return the subprocess return code as an integer orNoneif it hasn’t returned, which is similar to thesubprocess.Popen.returncode attribute.

SubprocessTransport.kill()¶

Kill the subprocess.

On POSIX systems, the function sends SIGKILL to the subprocess.On Windows, this method is an alias forterminate().

See alsosubprocess.Popen.kill().

SubprocessTransport.send_signal(signal)¶

Send thesignal number to the subprocess, as insubprocess.Popen.send_signal().

SubprocessTransport.terminate()¶

Stop the subprocess.

On POSIX systems, this method sendsSIGTERM to the subprocess.On Windows, the Windows API functionTerminateProcess() is called tostop the subprocess.

See alsosubprocess.Popen.terminate().

SubprocessTransport.close()¶

Kill the subprocess by calling thekill() method.

If the subprocess hasn’t returned yet, and close transports ofstdin,stdout, andstderr pipes.

Base Protocols¶

classasyncio.BaseProtocol¶

Base protocol with methods that all protocols share.

classasyncio.Protocol(BaseProtocol)¶

The base class for implementing streaming protocols(TCP, Unix sockets, etc).

classasyncio.BufferedProtocol(BaseProtocol)¶

A base class for implementing streaming protocols with manualcontrol of the receive buffer.

classasyncio.DatagramProtocol(BaseProtocol)¶

The base class for implementing datagram (UDP) protocols.

classasyncio.SubprocessProtocol(BaseProtocol)¶

The base class for implementing protocols communicating with childprocesses (unidirectional pipes).

Base Protocol¶

All asyncio protocols can implement Base Protocol callbacks.

Connection Callbacks

Connection callbacks are called on all protocols, exactly once pera successful connection. All other protocol callbacks can only becalled between those two methods.

BaseProtocol.connection_made(transport)¶

Called when a connection is made.

Thetransport argument is the transport representing theconnection. The protocol is responsible for storing the referenceto its transport.

BaseProtocol.connection_lost(exc)¶

Called when the connection is lost or closed.

The argument is either an exception object orNone.The latter means a regular EOF is received, or the connection wasaborted or closed by this side of the connection.

Flow Control Callbacks

Flow control callbacks can be called by transports to pause orresume writing performed by the protocol.

See the documentation of theset_write_buffer_limits()method for more details.

BaseProtocol.pause_writing()¶

Called when the transport’s buffer goes over the high watermark.

BaseProtocol.resume_writing()¶

Called when the transport’s buffer drains below the low watermark.

If the buffer size equals the high watermark,pause_writing() is not called: the buffer size mustgo strictly over.

Conversely,resume_writing() is called when thebuffer size is equal or lower than the low watermark. These endconditions are important to ensure that things go as expected wheneither mark is zero.

Streaming Protocols¶

Event methods, such asloop.create_server(),loop.create_unix_server(),loop.create_connection(),loop.create_unix_connection(),loop.connect_accepted_socket(),loop.connect_read_pipe(), andloop.connect_write_pipe()accept factories that return streaming protocols.

Protocol.data_received(data)¶

Called when some data is received.data is a non-empty bytesobject containing the incoming data.

Whether the data is buffered, chunked or reassembled depends onthe transport. In general, you shouldn’t rely on specific semanticsand instead make your parsing generic and flexible. However,data is always received in the correct order.

The method can be called an arbitrary number of times whilea connection is open.

However,protocol.eof_received()is called at most once. Onceeof_received() is called,data_received() is not called anymore.

Protocol.eof_received()¶

Called when the other end signals it won’t send any more data(for example by callingtransport.write_eof(), if the other end also usesasyncio).

This method may return a false value (includingNone), in which casethe transport will close itself. Conversely, if this method returns atrue value, the protocol used determines whether to close the transport.Since the default implementation returnsNone, it implicitly closes theconnection.

Some transports, including SSL, don’t support half-closed connections,in which case returning true from this method will result in the connectionbeing closed.

State machine:

start -> connection_made    [-> data_received]*    [-> eof_received]?-> connection_lost -> end

Buffered Streaming Protocols¶

Buffered Protocols can be used with any event loop methodthat supportsStreaming Protocols.

BufferedProtocol implementations allow explicit manual allocationand control of the receive buffer. Event loops can then use the bufferprovided by the protocol to avoid unnecessary data copies. Thiscan result in noticeable performance improvement for protocols thatreceive big amounts of data. Sophisticated protocol implementationscan significantly reduce the number of buffer allocations.

The following callbacks are called onBufferedProtocolinstances:

BufferedProtocol.get_buffer(sizehint)¶

Called to allocate a new receive buffer.

sizehint is the recommended minimum size for the returnedbuffer. It is acceptable to return smaller or larger buffersthan whatsizehint suggests. When set to -1, the buffer sizecan be arbitrary. It is an error to return a buffer with a zero size.

get_buffer() must return an object implementing thebuffer protocol.

BufferedProtocol.buffer_updated(nbytes)¶

Called when the buffer was updated with the received data.

nbytes is the total number of bytes that were written to the buffer.

BufferedProtocol.eof_received()¶

See the documentation of theprotocol.eof_received() method.

get_buffer() can be called an arbitrary numberof times during a connection. However,protocol.eof_received() is called at most onceand, if called,get_buffer() andbuffer_updated() won’t be called after it.

State machine:

start -> connection_made    [-> get_buffer        [-> buffer_updated]?    ]*    [-> eof_received]?-> connection_lost -> end

Datagram Protocols¶

Datagram Protocol instances should be constructed by protocolfactories passed to theloop.create_datagram_endpoint() method.

DatagramProtocol.datagram_received(data,addr)¶

Called when a datagram is received.data is a bytes object containingthe incoming data.addr is the address of the peer sending the data;the exact format depends on the transport.

DatagramProtocol.error_received(exc)¶

Called when a previous send or receive operation raises anOSError.exc is theOSError instance.

This method is called in rare conditions, when the transport (e.g. UDP)detects that a datagram could not be delivered to its recipient.In many conditions though, undeliverable datagrams will be silentlydropped.

Subprocess Protocols¶

Subprocess Protocol instances should be constructed by protocolfactories passed to theloop.subprocess_exec() andloop.subprocess_shell() methods.

SubprocessProtocol.pipe_data_received(fd,data)¶

Called when the child process writes data into its stdout or stderrpipe.

fd is the integer file descriptor of the pipe.

data is a non-empty bytes object containing the received data.

SubprocessProtocol.pipe_connection_lost(fd,exc)¶

Called when one of the pipes communicating with the child processis closed.

fd is the integer file descriptor that was closed.

SubprocessProtocol.process_exited()¶

Called when the child process has exited.

It can be called beforepipe_data_received() andpipe_connection_lost() methods.

TCP Echo Server¶

Create a TCP echo server using theloop.create_server() method, send backreceived data, and close the connection:

importasyncioclassEchoServerProtocol(asyncio.Protocol):defconnection_made(self,transport):peername=transport.get_extra_info('peername')print('Connection from{}'.format(peername))self.transport=transportdefdata_received(self,data):message=data.decode()print('Data received:{!r}'.format(message))print('Send:{!r}'.format(message))self.transport.write(data)print('Close the client socket')self.transport.close()asyncdefmain():# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()server=awaitloop.create_server(EchoServerProtocol,'127.0.0.1',8888)asyncwithserver:awaitserver.serve_forever()asyncio.run(main())

TCP Echo Client¶

A TCP echo client using theloop.create_connection() method, sendsdata, and waits until the connection is closed:

importasyncioclassEchoClientProtocol(asyncio.Protocol):def__init__(self,message,on_con_lost):self.message=messageself.on_con_lost=on_con_lostdefconnection_made(self,transport):transport.write(self.message.encode())print('Data sent:{!r}'.format(self.message))defdata_received(self,data):print('Data received:{!r}'.format(data.decode()))defconnection_lost(self,exc):print('The server closed the connection')self.on_con_lost.set_result(True)asyncdefmain():# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()on_con_lost=loop.create_future()message='Hello World!'transport,protocol=awaitloop.create_connection(lambda:EchoClientProtocol(message,on_con_lost),'127.0.0.1',8888)# Wait until the protocol signals that the connection# is lost and close the transport.try:awaiton_con_lostfinally:transport.close()asyncio.run(main())

UDP Echo Server¶

A UDP echo server, using theloop.create_datagram_endpoint()method, sends back received data:

importasyncioclassEchoServerProtocol:defconnection_made(self,transport):self.transport=transportdefdatagram_received(self,data,addr):message=data.decode()print('Received%r from%s'%(message,addr))print('Send%r to%s'%(message,addr))self.transport.sendto(data,addr)asyncdefmain():print("Starting UDP server")# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()# One protocol instance will be created to serve all# client requests.transport,protocol=awaitloop.create_datagram_endpoint(EchoServerProtocol,local_addr=('127.0.0.1',9999))try:awaitasyncio.sleep(3600)# Serve for 1 hour.finally:transport.close()asyncio.run(main())

UDP Echo Client¶

A UDP echo client, using theloop.create_datagram_endpoint()method, sends data and closes the transport when it receives the answer:

importasyncioclassEchoClientProtocol:def__init__(self,message,on_con_lost):self.message=messageself.on_con_lost=on_con_lostself.transport=Nonedefconnection_made(self,transport):self.transport=transportprint('Send:',self.message)self.transport.sendto(self.message.encode())defdatagram_received(self,data,addr):print("Received:",data.decode())print("Close the socket")self.transport.close()deferror_received(self,exc):print('Error received:',exc)defconnection_lost(self,exc):print("Connection closed")self.on_con_lost.set_result(True)asyncdefmain():# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()on_con_lost=loop.create_future()message="Hello World!"transport,protocol=awaitloop.create_datagram_endpoint(lambda:EchoClientProtocol(message,on_con_lost),remote_addr=('127.0.0.1',9999))try:awaiton_con_lostfinally:transport.close()asyncio.run(main())

Connecting Existing Sockets¶

Wait until a socket receives data using theloop.create_connection() method with a protocol:

importasyncioimportsocketclassMyProtocol(asyncio.Protocol):def__init__(self,on_con_lost):self.transport=Noneself.on_con_lost=on_con_lostdefconnection_made(self,transport):self.transport=transportdefdata_received(self,data):print("Received:",data.decode())# We are done: close the transport;# connection_lost() will be called automatically.self.transport.close()defconnection_lost(self,exc):# The socket has been closedself.on_con_lost.set_result(True)asyncdefmain():# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()on_con_lost=loop.create_future()# Create a pair of connected socketsrsock,wsock=socket.socketpair()# Register the socket to wait for data.transport,protocol=awaitloop.create_connection(lambda:MyProtocol(on_con_lost),sock=rsock)# Simulate the reception of data from the network.loop.call_soon(wsock.send,'abc'.encode())try:awaitprotocol.on_con_lostfinally:transport.close()wsock.close()asyncio.run(main())

loop.subprocess_exec() and SubprocessProtocol¶

An example of a subprocess protocol used to get the output of asubprocess and to wait for the subprocess exit.

The subprocess is created by theloop.subprocess_exec() method:

importasyncioimportsysclassDateProtocol(asyncio.SubprocessProtocol):def__init__(self,exit_future):self.exit_future=exit_futureself.output=bytearray()self.pipe_closed=Falseself.exited=Falsedefpipe_connection_lost(self,fd,exc):self.pipe_closed=Trueself.check_for_exit()defpipe_data_received(self,fd,data):self.output.extend(data)defprocess_exited(self):self.exited=True# process_exited() method can be called before# pipe_connection_lost() method: wait until both methods are# called.self.check_for_exit()defcheck_for_exit(self):ifself.pipe_closedandself.exited:self.exit_future.set_result(True)asyncdefget_date():# Get a reference to the event loop as we plan to use# low-level APIs.loop=asyncio.get_running_loop()code='import datetime; print(datetime.datetime.now())'exit_future=asyncio.Future(loop=loop)# Create the subprocess controlled by DateProtocol;# redirect the standard output into a pipe.transport,protocol=awaitloop.subprocess_exec(lambda:DateProtocol(exit_future),sys.executable,'-c',code,stdin=None,stderr=None)# Wait for the subprocess exit using the process_exited()# method of the protocol.awaitexit_future# Close the stdout pipe.transport.close()# Read the output which was collected by the# pipe_data_received() method of the protocol.data=bytes(protocol.output)returndata.decode('ascii').rstrip()date=asyncio.run(get_date())print(f"Current date:{date}")

See also thesame examplewritten using high-level APIs.