AsyncAPY - Code Examples

Filters Examples

Here is an example on how to use AsyncAPY’s Filters objects

from asyncapy import Server
from asyncapy.filters import Filters

server = Server(addr="0.0.0.0", port=1500)

# This filter will match any digit in the 'foo' field,
# and anything in the 'bar' field, e.g.:
# {"foo": 12355, "bar": "anything"}


@server.add_handler(Filters.Fields(foo="\d+", bar=None))
async def filtered_handler(client, packet):
    print(f"Look at this! {client} sent me {packet}!")
    await client.close()


server.start()

You can also use multiple Filters objects, by doing the following:

from asyncapy import Server
from asyncapy.filters import Filters

server = Server(addr="0.0.0.0", port=1500)

# This filter will match any digit in the 'foo' field,
# and anything in the 'bar' field, e.g.:
# {"foo": 12355, "bar": "123lmao"}
# Also, only packets coming from localhost (127.0.0.1) and from 151.53.88.15, will reach
# this handler


@server.add_handler(
    Filters.Fields(foo=r"\d+", bar=None), Filters.Ip(["127.0.0.1", "151.53.88.15"])
)
async def filtered_handler(client, packet):
    # code here
    ...


server.start()

You can pass as many filters as you want in any order. For a detailed look at filters check their docs.

Groups Examples

from asyncapy import Server

server = Server(addr="127.0.0.1", port=1500)


@server.add_handler()
async def echo_server(client, packet):
    print(f"Hello world from {client}!")
    print(f"Echoing back {packet}...")
    await client.send(packet)
    await client.close()


@server.add_handler(group=-1)
async def echo_server_2(client, packet):
    print(f"Hello world from {client} inside a group!")
    print(f"Echoing back {packet}...")
    await client.send(packet)
    # packet.stop_propagation()  # This would prevent the packet from being forwarded to the next handler


server.start()