irc3.plugins.asynchronious Asynchronious events

This module provide a way to catch data from various predefined events.

Usage

You’ll have to define a subclass of AsyncEvents:

class Whois(AsyncEvents):

    # the command will fail if we do not have a result after 30s
    timeout = 20

    # send this line before listening to events
    send_line = 'WHOIS {nick} {nick}'

    # when those events occurs, we can add them to the result list
    events = (
        # (?i) is for IGNORECASE. This will match either NicK or nick
        {'match': "(?i)^:\S+ 301 \S+ {nick} :(?P<away>.*)"},
        {'match': "(?i)^:\S+ 311 \S+ {nick} (?P<username>\S+) (?P<host>\S+) . "
                  ":(?P<realname>.*)(?i)"},
        {'match': "(?i)^:\S+ 312 \S+ {nick} (?P<server>\S+) "
                  ":(?P<server_desc>.*)"},
        {'match': "(?i)^:\S+ 317 \S+ {nick} (?P<idle>[0-9]+).*"},
        {'match': "(?i)^:\S+ 319 \S+ {nick} :(?P<channels>.*)", 'multi': True},
        {'match': "(?i)^:\S+ 330 \S+ {nick} (?P<account>\S+) "
                  ":(?P<account_desc>.*)"},
        {'match': "(?i)^:\S+ 671 \S+ {nick} :(?P<connection>.*)"},
        # if final=True then a result is returned when the event occurs
        {'match': "(?i)^:\S+ (?P<retcode>(318|401)) \S+ (?P<nick>{nick}) :.*",
         'final': True},
    )

    def process_results(self, results=None, **value):
        """take results list of all events and put them in a dict"""
        channels = []
        for res in results:
            channels.extend(res.pop('channels', '').split())
            value.update(res)
        value['channels'] = channels
        value['success'] = value.get('retcode') == '318'
        return value

Notice that regexps and send_line contains some {nick}. This will be substitued later with the keyword arguments passed to the instance.

Then you’re able to use it in a plugin:

class MyPlugin:

    def __init__(self, bot):
        self.bot = bot
        self.whois = Whois(bot)

    def do_whois(self):
        # remember {nick} in the regexp? Here it is
        whois = await self.whois(nick='gawel')
        if int(whois['idle']) / 60 > 10:
            self.bot.privmsg('gawel', 'Wake up dude')

Warning

Your code should always check if the result has been set before timeout by using result[‘timeout’] which is True when the bot failed to get a result before 30s (you can override the default value per call)

Warning

Do not over use this feature. If you’re making a lot of calls at the same time you should experience some weird behavior since irc do not allow to identify responses for a command. That’s why the exemple use {nick} in the regexp to filter events efficiently. But two concurent call for the same nick can still fail.

API

class irc3.asynchronous.AsyncEvents(context)

Asynchronious events

__call__(**kwargs)

Register events; and callbacks then return a asyncio.Future. Events regexp are compiled with params

process_results(results=None, **value)

Process results. results is a list of dict catched during event. value is a dict containing some metadata (like timeout=(True/False).

class irc3.plugins.asynchronious.Async(context)

Asynchronious plugin. Extend the bot with some common commands using AsyncEvents

async_who_channel_flags(channel, flags, timeout)

Creates and calls a class from WhoChannelFlags with needed match rule for WHO command on channels with flags.

channel_bans(channel, timeout=20)

Send a MODE +b and return a Future which will contain recieved data: .. code-block:: py

result = await bot.async_cmds.channel_bans(‘#irc3’)

ctcp_async(nick, ctcp, timeout=20)

Send a CTCP and return a Future which will contain recieved data:

result = await bot.async_cmds.ctcp('irc3', 'version')

ison(*nicknames, **kwargs)

Send a ISON and return a Future which will contain recieved data:

result = await bot.async_cmds.ison('gawel', 'irc3')

names(channel, timeout=20)

Send a NAMES and return a Future which will contain recieved data:

result = await bot.async_cmds.names('#irc3')

who(target, flags=None, timeout=20)

Send a WHO and return a Future which will contain recieved data:

result = await bot.async_cmds.who('gawel')
result = await bot.async_cmds.who('#irc3')
result = await bot.async_cmds.who('#irc3', 'an')
# or
result = await bot.async_cmds.who('#irc3', ['a', 'n'])

whois(nick, timeout=20)

Send a WHOIS and return a Future which will contain recieved data:

result = await bot.async_cmds.whois('gawel')