Event Reference

gd.py provides a system for handling events, such as new daily levels or weekly demons.

Example

Here is a simple usage example:

import gd

client = gd.Client()

@client.listen_for('daily')  # listen_for returns a client.event decorator
async def on_new_daily(level):
    print('New daily level was set: {!r}.'.format(level))

gd.events.start()

This code snippet will print a message every time a new level is getting daily.

Example Explanation

client.listen_for('event_type') creates a listener with the client, and schedules the loop task.

gd.events.start() runs all the scheduled tasks in another thread.

Warning

Do not call client.listen_for after gd.events.start() was called, since it simply will not have any effect and will not start another listener.

Built-In Listeners

name	event callback	listener class
daily	Client.on_new_daily()	events.TimelyLevelListener
weekly	Client.on_new_weekly()	events.TimelyLevelListener
rate	Client.on_level_rated()	events.RateLevelListener
unrate	Client.on_level_unrated()	events.RateLevelListener
friend_request	Client.on_friend_request()	events.MessageOrRequestListener
message	Client.on_message()	events.MessageOrRequestListener
level_comment	Client.on_level_comment()	events.LevelCommentListener

class gd.events.TimelyLevelListener(client: gd.client.Client, timely_type: str, delay: int = 10.0)

Listens for a new daily or weekly level.

class gd.events.RateLevelListener(client: gd.client.Client, listen_to_rate: bool = True, delay: float = 10.0)

Listens for a new rated or unrated level.

class gd.events.MessageOrRequestListener(client: gd.client.Client, listen_messages: bool = True, delay: float = 5.0)

Listens for a new friend request or message.

class gd.events.LevelCommentListener(client: gd.client.Client, level_id: int, delay: float = 10.0)

Listens for a new comment on a level.

Running Manually

If you wish to run the listener normally (blocking the main thread), you can do the following:

import gd

loop = gd.utils.acquire_loop()

client = gd.Client()
client.listen_for('daily')

@client.event
async def on_new_daily(level):
    print(level.creator.name)

gd.events.run(loop)

Instead of running this:

gd.events.run(loop)

You can use this:

gd.events.enable(loop)
loop.run_forever()

This allows for using gd.py, for example, in the discord.py bot:

from discord.ext import commands
import gd

bot = commands.Bot(command_prefix="!")
client = gd.Client()
bot.client = client  # attach the client in case you might need to use it

@client.listen_for("daily")
async def on_new_daily(level: gd.Level) -> None:
    ...  # you can do something with your bot here

gd.events.enable(bot.loop)
bot.run("BOT_TOKEN")

Event handlers with @client.event

As shown in examples above, new implementation for an event can be registered with @client.event decorator. See Client.event() for more info.

Event handlers with subclassing gd.Client

Another way to write an implementation for on_event task is to subclass Client.

import gd

class MyClient(gd.Client):
    async def on_new_daily(self, level: gd.Level) -> None:
        print(level)

client = MyClient()

client.listen_for('daily')

gd.events.start()

Functions

gd.events.enable(loop: Optional[asyncio.events.AbstractEventLoop] = None) → None

Attach all listeners to the loop. If not provided, default loop is used.

gd.events.run(loop: Optional[asyncio.events.AbstractEventLoop] = None) → None

Run listeners in a given loop.

gd.events.start(loop: Optional[asyncio.events.AbstractEventLoop] = None) → None

Does the same as run(), but in the remote thread.

gd.events.disable() → None

Shorthand for calling all three functions:

gd.events.cancel_tasks()
gd.events.shutdown_loops()
gd.events.shutdown_threads()

This function is intended to be called at the end of programs, and is also run on shutdown.

gd.events.cancel_tasks() → None

Cancel all running task-loop objects.

gd.events.attach_to_loop(loop: asyncio.events.AbstractEventLoop) → None

Set loop as default loop in gd.events.

Creating Custom Listeners

It is possible to implement your own listeners.

The main idea is subclassing AbstractListener and creating your own scan method in there.

class gd.events.AbstractListener(client: gd.client.Client, delay: float = 10.0)

Abstract listener for listeners to derive from.

await on_error(exc: Exception) → None

Basic event handler to print the errors if any occur.

await setup() → None

This function is used to do some preparations before starting listeners.

await scan() → None

This function should contain main code of the listener.

await main() → None

Main function, that is basically doing all the job.

import gd

class CustomListener(gd.events.AbstractListener):
    def __init__(self, client, delay=10.0):
        # you can add additional arguments here
        super().__init__(client=client, delay=delay)  # this line is required

    async def scan(self):
        # do something here
        ...
        # dispatch event
        dispatcher = self.client.dispatch('event', *args, **kwargs)
        # dispatches on_<event> with args and kwargs
        self.loop.create_task(dispatcher)  # schedule execution

client.listener.append(CustomListener(client, 5.0))

gd.events.start()