[SOLVED] Why is this asyncio task blocking?

[SecureCoop]

This is my first time working with asyncio and I can't understand why this task is blocking. Everything is awaited that should be awaited. Please help? Using Python 3.10.6 on Ubuntu 22.04.2 LTS.

#!/usr/bin/env python
import ssl
import asyncio
from aiomysql import create_pool, DictCursor
from config import conf


user = conf['MySQL SecureCoop user']
password = conf['MySQL SecureCoop password']
host = conf['MySQL host']
# TODO Move into config
db = 'securecoop'
cafile = '/etc/nagios/ssl/cacert.pem'


async def connect():
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ssl_ctx.verify_mode = ssl.CERT_REQUIRED
    ssl_ctx.check_hostname = False
    ssl_ctx.load_verify_locations(cafile=cafile)

    # TODO Move into config
    init_command = """\
        SET autocommit = 1;
        SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
        """
    # TODO Move into config
    autocommit = True

    loop = asyncio.get_event_loop()
    # TODO Move into config
    maxsize = 100

    pool = await create_pool(user=user,
                             db=db,
                             host=host,
                             password=password,
                             loop=loop,
                             ssl=ssl_ctx,
                             init_command=init_command,
                             maxsize=maxsize,
                             autocommit=autocommit)
    return pool


async def query(sql, params: tuple = tuple()):
    pool = await connect()
    async with pool.acquire() as conn:
        cur = await conn.cursor(DictCursor)
        await cur.execute(sql, params)
        return cur


async def outgoing_worker():
    while True:
        sql = 'SELECT * FROM `websocket_outgoing_queue`'
        results = await query(sql)
        print(repr(results))

        for row in await results.fetchall():
            serial = row.get('serial', None)

            sql = '''DELETE FROM `websocket_outgoing_queue`
            WHERE `serial` = %s'''
            params = (serial,)
            await query(sql, params)

            values = row.get('values', None)

            print(f'ws.send(json.dumps({repr(values)}))')
        print('About to await asyncio.sleep(1)')
        await asyncio.sleep(1)


async def asyncio_create_queue_tasks():
    outgoing_worker_task = asyncio.create_task(outgoing_worker())
    await outgoing_worker_task
    print('Should reach here inside asyncio_create_queue_tasks()')


loop = asyncio.get_event_loop()
print('About to asyncio_create_queue_tasks()')
asyncio.run(asyncio_create_queue_tasks())
print('Should reach here inside the main routine')
loop.run_forever()

Output:
(SecureCoop) root@api02:/SecureCoop# ./test_asyncio_db_async.py
/SecureCoop/./test_asyncio_db_async.py:81: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
About to asyncio_create_queue_tasks()
<aiomysql.cursors.DictCursor object at 0x7fe1ecd88e50>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7fe1ecd88f10>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7fe1ecd8a500>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7fe1ecd8b220>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7fe1ecd88fa0>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7fe1ecd89f30>
About to await asyncio.sleep(1)
...

[SecureCoop]

Found the bug. Need to use loop.create_task(). Modified code:

#!/usr/bin/env python
import ssl
import asyncio
from aiomysql import create_pool, DictCursor
from config import conf


user = conf['MySQL SecureCoop user']
password = conf['MySQL SecureCoop password']
host = conf['MySQL host']
# TODO Move into config
db = 'securecoop'
cafile = '/etc/nagios/ssl/cacert.pem'


async def connect():
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ssl_ctx.verify_mode = ssl.CERT_REQUIRED
    ssl_ctx.check_hostname = False
    ssl_ctx.load_verify_locations(cafile=cafile)

    # TODO Move into config
    init_command = """\
        SET autocommit = 1;
        SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
        """
    # TODO Move into config
    autocommit = True

    loop = asyncio.get_event_loop()
    # TODO Move into config
    maxsize = 100

    pool = await create_pool(user=user,
                             db=db,
                             host=host,
                             password=password,
                             loop=loop,
                             ssl=ssl_ctx,
                             init_command=init_command,
                             maxsize=maxsize,
                             autocommit=autocommit)
    return pool


async def query(sql, params: tuple = tuple()):
    pool = await connect()
    async with pool.acquire() as conn:
        cur = await conn.cursor(DictCursor)
        await cur.execute(sql, params)
        return cur


async def outgoing_worker():
    while True:
        sql = 'SELECT * FROM `websocket_outgoing_queue`'
        results = await query(sql)
        print(repr(results))

        for row in await results.fetchall():
            serial = row.get('serial', None)

            sql = '''DELETE FROM `websocket_outgoing_queue`
            WHERE `serial` = %s'''
            params = (serial,)
            await query(sql, params)

            values = row.get('values', None)

            print(f'ws.send(json.dumps({repr(values)}))')
        print('About to await asyncio.sleep(1)')
        await asyncio.sleep(1)


loop = asyncio.get_event_loop()
loop.create_task(outgoing_worker())
print('Should reach here inside the main routine')
loop.run_forever()

I can see the output below says "Should reach here inside the main routine" as I want it to.

Output:
(SecureCoop) root@api02:/SecureCoop# ./test_asyncio_db_async.py
/SecureCoop/./test_asyncio_db_async.py:75: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
Should reach here inside the main routine
<aiomysql.cursors.DictCursor object at 0x7f727255cb20>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7f727255cb80>
About to await asyncio.sleep(1)
<aiomysql.cursors.DictCursor object at 0x7f727255e140>
About to await asyncio.sleep(1)
...