Asynchronous Clients#

Added in version 0.5.0.

This covers the asynchronous I/O client provided by little-a2s.

little-a2s only provides an asyncio client. If you need to integrate with a different event loop such as Trio, anyio, or gevent, you can use the Sans-I/O interface to build your own client.

class little_a2s.AsyncA2S(connector)#

Bases: DatagramProtocol

An asynchronous client for A2S queries.

a2s = AsyncA2S.from_addr("127.0.0.1", 27015)
async with a2s, asyncio.timeout(1):
    print(await a2s.info())
    print(await a2s.players())
    print(await a2s.rules())

This follows the Source format. For the Goldsource equivalent, see AsyncA2SGoldsource.

This class supports the asynchronous context manager protocol which calls the connector function to set the transport + remote address, and closes the transport upon exit.

Parameters:: connector (Callable[[Self], Awaitable[tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes] | None]]) – The function to call and await to create a datagram transport and return the remote address, if any. See also from_addr(), from_ipv4(), and from_ipv6().

Added in version 0.5.0.

classmethod from_addr(host, port, *, prefer_ipv4=True)#

Resolve the given host and create an A2S query.

Parameters:

host (str) – The IPv4 address, IPv6 address, or domain name to query.
port (int) – The port to query.
prefer_ipv4 (bool) – If True, prefer to resolve hostnames to IPv4 addresses.

Return type:

Self

classmethod from_ipv4()#

Create an A2S query with a UDP IPv4 socket not connected to any address.

This allows you to use the same socket with addr arguments:

async with AsyncA2S.from_ipv4() as a2s, asyncio.timeout(1):
    info = await a2s.info(("127.0.0.1", 2303))
    info = await a2s.info(("127.0.0.1", 27015))

Return type:: Self

classmethod from_ipv6()#

Create an A2S query with a UDP IPv6 socket not connected to any address.

This allows you to use the same socket with addr arguments:

async with AsyncA2S.from_ipv6() as a2s, asyncio.timeout(1):
    info = await a2s.info(("::1", 2303))
    info = await a2s.info(("::1", 27015))

Return type:: Self

async close()#

Close the current datagram transport, raising any exception if the connection improperly closed.

Raises:: RuntimeError – The transport is not connected.
Return type:: None

async info(addr=None)#

Send an A2S_INFO request and wait for a response.

Parameters:

addr (tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes] | None) – The address to send the request to. Does not apply if socket is already connected to an address, such as from from_addr().

Raises: