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.

Changed in version 0.7.0: Fixed PayloadError not being raised by request methods.

Concurrent, pending requests to one address can raise the same PayloadError if any received packet is malformed.

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:
Return type:

ClientEventInfo

async players(addr=None)#

Send an A2S_PLAYER 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:
Return type:

ClientEventPlayers

async rules(addr=None)#

Send an A2S_RULES 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:
Return type:

ClientEventRules

async start()#

Call the connector function to create the datagram transport.

Raises:
  • OSError – The address could not be resolved.

  • RuntimeError – The transport is already connected.

Return type:

None

property transport: DatagramTransport#

The current datagram transport.

Raises:

RuntimeError – The transport is not connected.

class little_a2s.AsyncA2SGoldsource(connector)#

Bases: AsyncA2S

A asynchronous client for A2S Goldsource queries.

Parameters:

connector (Callable[[Self], Awaitable[tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes] | 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:
Return type:

ClientEventGoldsourceInfo