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 decorator
async def on_new_daily(level):
    print(f"New daily level was set: {level.name} (ID: {level.id}).")

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	on_new_daily()	TimelyLevelListener
weekly	on_new_weekly()	TimelyLevelListener
rate	on_level_rated()	RateLevelListener
unrate	on_level_unrated()	RateLevelListener
friend_request	on_friend_request()	MessageOrRequestListener
message	on_message()	MessageOrRequestListener
level_comment	on_level_comment()	LevelCommentListener

class gd.events.TimelyLevelListener(client: Client, daily: bool, delay: float = 10.0)

Listens for new daily or weekly levels.

class gd.events.RateLevelListener(client: Client, rate: bool, delay: float = 10.0, pages: int = 5, ensure: bool = True)

Listens for new rated or unrated levels.

class gd.events.MessageOrRequestListener(client: Client, message: bool, delay: float = 10.0, pages: int = 5, read: bool = True)

Listens for new friend requests or messages.

class gd.events.LevelCommentListener(level_id: int, client: Client, delay: float = 10.0, amount: int = 1000, refresh: bool = True)

Listens for new comments on given 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 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.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: Client, delay: float = 10.0)

Abstract listener for listeners to derive from.

async on_error(error: Exception) → None

Basic event handler to print the errors if any occur.

async setup() → None

This function is used to do some preparations before running scan iteration.

async scan() → None

This function should contain main code of the listener.

async 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)  # call listener init

    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()