Asyncio in loop waiting for event

Question

My problem is that I have script for event listening. The event listening start in loop after connection to the server. So after server connection in function main I have create loop

loop = asyncio.get_event_loop()
        asyncio.ensure_future(monitor(ts3conn, dbconn))
        loop.run_forever()

Now when I get any event, check if statements, if there is True, I have to await function or create new task? I want to async 3 functions, 1 main, listening all the time and additional what will be created when someone write message/join channel, in this script asyncio still not working, when 30 users join channel 777 (if 777 == int(event['ctid']):) and in meantime someone will join channel 904 (if 904 == int(event['ctid']):), last guy must wait untill that 30 users will be served (I hope you understand)

My code:

import ts3
import time
import logging
import json
import pymysql.cursors
import pymysql
import asyncio
from logging.handlers import RotatingFileHandler

def main():
    with ts3.query.TS3ServerConnection(URI) as ts3conn:
        # connect to server instance, update name and go to specific channel
        ts3conn.exec_("use", sid=SID)
        ts3conn.exec_("clientupdate", client_nickname=CLIENT_NAME)
        myclid = ts3conn.exec_("whoami")[0]["client_id"]
        ts3conn.exec_("clientmove", clid=myclid, cid=JOIN_CHANNEL_ID)
        ts3conn.exec_("servernotifyregister", event="server")
        ts3conn.exec_("servernotifyregister", event="channel", id=0)
        ts3conn.exec_("servernotifyregister", event="textprivate")
        dbconn = pymysql.connect(host='localhost', user='root', password='', db='teamspeak')

        loop = asyncio.get_event_loop()
        asyncio.ensure_future(monitor(ts3conn, dbconn))
        loop.run_forever()

# Function handling the events and initiating activity logs
async def monitor(ts3conn, dbconn):
    # register for all events in server wide chat
    ts3conn.exec_("servernotifyregister", event="server")
    ts3conn.exec_("servernotifyregister", event="channel", id=0)
    ts3conn.exec_("servernotifyregister", event="textprivate")
    ts3conn.send_keepalive()
    while True:
        try:
            event = ts3conn.wait_for_event(timeout=10)[0]
        except ts3.query.TS3TimeoutError:
            ts3conn.send_keepalive()
        else:
            await asyncio.sleep(0.001)
            print(event)
            # ============= IF JOIN CHANNEL ===================
            if "ctid" in event.keys() and "clid" in event.keys() and int(event['ctid']) != 0:
                if 777 == int(event['ctid']):
                    asyncio.create_task(first(ts3conn, dbconn, event['clid']))
                    #ts3conn.exec_("clientkick", reasonid=4, clid=event['clid'])
                if 904 == int(event['ctid']):
                    asyncio.create_task(second(ts3conn, dbconn, event['clid']))
                    #ts3conn.exec_("clientkick", reasonid=4, clid=event['clid'])
            # ============= IF SEND MSG ===================
            if "msg" in event.keys() and "invokeruid" in event.keys() and 'serveradmin' not in str(event['invokeruid']):
                if event['msg'] == "!info":
                    print("info")
                    asyncio.create_task(first(ts3conn, dbconn, event['invokerid']))


async def first(ts3conn, dbconn, uid):
    try:
        print("first")
        user = ts3conn.exec_("clientinfo", clid=uid)
        if any(i in user[0]['client_servergroups'] for i in REG):
            try:
                sql = "SELECT * FROM users WHERE uid=%s"
                cursor = dbconn.cursor()
                cursor.execute(sql, (user[0]['client_unique_identifier']))
                c = cursor.fetchone()
                ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"register: {c}")
            except KeyError as e:
                print(e)
        else:
            ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"not register")
    except KeyError as e:
        print(f"keyerror: {e}")


async def second(ts3conn, dbconn, uid):
    try:
        user = ts3conn.exec_("clientinfo", clid=uid)
        if any(i in user[0]['client_servergroups'] for i in REG):
            try:
                sql = "SELECT * FROM users WHERE uid=%s"
                cursor = dbconn.cursor()
                cursor.execute(sql, (user[0]['client_unique_identifier']))
                c = cursor.fetchone()
                ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"1 out: {c}")
            except KeyError as e:
                print(e)
        else:
            ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"321123321321132")
    except KeyError as e:
        print(f"keyerror: {e}")



if __name__ == "__main__":
    with open('config.json') as config_file:
        config = json.load(config_file)

    try:
        SQLDATABASE = config["sqldatabase"]
        DATABASE = config["sqldatabase"]
        URI = config["uri"]
        SID = config["sid"]
        CLIENT_NAME = config["client_name"]
        JOIN_CHANNEL_ID = config["join_channel_id"]
        REG = config['zarejeya']

        if config["log_level"] == "CRITICAL":
            LOG_LEVEL = logging.CRITICAL
        elif config["log_level"] == "ERROR":
            LOG_LEVEL = logging.ERROR
        elif config["log_level"] == "WARNING":
            LOG_LEVEL = logging.WARNING
        elif config["log_level"] == "INFO":
            LOG_LEVEL = logging.INFO
        elif config["log_level"] == "DEBUG":
            LOG_LEVEL = logging.DEBUG
        else:
            LOG_LEVEL = logging.NOTSET
    except:
        print("Error parsing config")
        raise

    log_formatter = logging.Formatter("%(asctime)s - %(funcName)s - %(levelname)s - %(message)s")
    log_handler = RotatingFileHandler("ts3bot.log", mode='a', maxBytes=50 * 1024 * 1024, backupCount=2)
    log_handler.setFormatter(log_formatter)
    log_handler.setLevel(LOG_LEVEL)
    # noinspection PyRedeclaration
    logger = logging.getLogger("root")
    logger.setLevel(LOG_LEVEL)
    logger.addHandler(log_handler)

    while True:
        try:
            main()
        except Exception:
            logger.exception("Exception occurred and connection is closed")
            logger.info("Trying to restart in 30s")
            time.sleep(30)

I need something like this discord bot: https://tutorials.botsfloor.com/a-discord-bot-with-asyncio-359a2c99e256 but I cant make it here...

similar: https://stackoverflow.com/questions/57578794/why-is-reading-and-calling-an-api-from-a-file-slower-using-python-async-than-syn/57579373#57579373 — Sam Daniel, Sep 01 '19 at 16:05

score 0 · Answer 1 · answered Sep 01 '19 at 15:52

0

You can use loop.create_task() to run async methods in the background without blocking the current loop.

You can read more on that in the official docs. https://docs.python.org/3/library/asyncio-task.html

answered Sep 01 '19 at 15:52

painor

1,119
1
8
25

score 0 · Accepted Answer · answered Sep 01 '19 at 16:40

0

Calling await() will put the current coroutine in suspension and will only resume when the awaited coro is returned.

In your code, monitor coroutine will process the incoming events serially.. So, if any event occurs, the monitor will await on the first()/second() coros, which will put the monitor coro again in suspension... So this is synchronous.. No real benefit of async.. instead it makes the performance worse..

The event handling coroutines, first() and second() can be converted into tasks.. You can use asyncio.create_task() to wrap a coroutine to a task.. This will be scheduled to execute in the same event loop when it is its turn. and the calling coro(int this case, monitor()) will not be suspended while the task is executing.. The task itself is an awaitable, so calling await on a task will put the callee to suspension.

Perhaps, you might want to save created task (returned during create_task()) for later operations.. like stopping it, awaiting it..

await task # To await for the task to complete, suspends to current coro.

If you wish to cancel an active task, then the task will throw an CancelledError, which has to be handled neatly.

Example:

async def stop_task(task):
    if not task:
        return
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

answered Sep 01 '19 at 16:40

Sam Daniel

1,800
12
22

When I add asyncio.create_task to first and second, It looks like function main work asynchronous but functions first and second wont execute, I dont have to wait for function first or second, I want to let it make in background without getting any results – Mark Loki Sep 01 '19 at 17:02
Coroutines are non-preemptable methods.. They have to voluntarily signal that they can be put on hold.. And await facilitates that.. if ts3conn.wait_for_event is not an await statement, then try to use await asyncio.sleep(0.01) after creating the task. This way it puts the monitor to suspend and gives space for first()/second() to execute – Sam Daniel Sep 01 '19 at 17:10
After read your message I'v already do that, I have add await asyncio.sleep(0.001) after ```while True: try: event = ts3conn.wait_for_event(timeout=10)[0] except ts3.query.TS3TimeoutError: ts3conn.send_keepalive() else:``` And.... When user join channel 777, nothing happen, when he left, message was send... I think, I need put await somewhere else but where... I have add new code at the top – Mark Loki Sep 01 '19 at 17:12
shouldn't you add await asyncio.sleep() inside the while loop? and not after while True.. – Sam Daniel Sep 01 '19 at 17:15
obviously, my bad, thank you for help, It works now :) – Mark Loki Sep 01 '19 at 17:17
Not at all, I think it is not working properly, it looks like it still execute functions one by one, the script start working when someone join channels etc, but if 30 users join channel 777 and in meantime someone join channel 904 or send message !info to bot he must wait untill script stop sending messages to that 30 ppls, does asyncio can help in that case? – Mark Loki Sep 01 '19 at 17:42
Asyncio is not parallel processing like multi threading. For an event loop there is always only one task that is executing at any point of time. The advantage is you can do other tasks when you are really blocked for an IO or some blocking call. If 30 users join, then 30 tasks are create and the next event will be queued after this. – Sam Daniel Sep 01 '19 at 17:51

Asyncio in loop waiting for event

2 Answers2