import asyncio
import ctypes
import struct
import pyngres.asyncio as py
...

DBEVENT_POLLING_INTERVAL = 0.1  # seconds

connHandle = None
tranHandle = None
...

# instantiate the required asyncio controls
run_flag = asyncio.Event()
session_mutex = asyncio.Lock()
event_queue = asyncio.Queue()
...

async def dbevent_catcher():
    '''task to catch DBEvents'''

    # a DBEvent can include an optional message; prepare to buffer it
    gdp = py.IIAPI_GETDESCRPARM()
    event_msg = ctypes.c_buffer(258)
    event_value = py.IIAPI_DATAVALUE()
    event_value.dv_length = len(event_msg)
    event_value.dv_value = ctypes.addressof(event_msg)
    eventData = (py.IIAPI_DATAVALUE * 1)()
    eventData[0] = event_value
    gcp = py.IIAPI_GETCOLPARM()
    gcp.gc_rowCount = 1
    gcp.gc_columnCount = 1
    gcp.gc_columnData = eventData        

    # get a DBEvent notification eventHandle
    cep = py.IIAPI_CATCHEVENTPARM()
    cep.ce_connHandle = connHandle
    async with session_mutex:
        py.IIapi_catchEvent(cep)
        # the eventHandle is returned immediately
        eventHandle = cep.ce_eventHandle

    # busy-wait loop to check for notification
    while run_flag.is_set():
        if cep.ce_genParm.gp_completed:
            # capture the DBEvent description and queue it
            event = dict()
            event['eventDB'] = cep.ce_eventDB[:] 
            event['eventName'] = cep.ce_eventName[:]
            event['eventOwner'] = cep.ce_eventOwner[:]
            date = cep.ce_eventTime.dv_value
            event['eventTime'] = ingresdate_to_str(date)
            if cep.ce_eventInfoAvail:
                # fetch the event message
                async with session_mutex:
                    gdp.gd_stmtHandle = eventHandle
                    await py.IIapi_getDescriptor(gdp)
                    gcp.gc_stmtHandle = eventHandle                        
                    await py.IIapi_getColumns(gcp)
                    # the message is returned as a VARCHAR so unpack it
                    msg_length,msg_bytes= struct.unpack('h256s',event_msg)
                    msg = msg_bytes[:msg_length].decode()
                    event['eventText'] = msg
            else:
                event['eventText'] = None
            # put the DBEvent notification in the global queue
            await event_queue.put(event)

            # wait for the next notification, reusing the event handle
            cep = py.IIAPI_CATCHEVENTPARM()
            cep.ce_connHandle = connHandle
            cep.ce_eventHandle = eventHandle
            async with session_mutex:
                py.IIapi_catchEvent(cep)
            # surrender control back to asyncio
            await asyncio.sleep(0)
        else:
            # no event yet; sleep through a polling interval 
            await asyncio.sleep(DBEVENT_POLLING_INTERVAL)

async def dbevent_getter():
    '''task to solicit DBEvent notifications'''

    # the OpenAPI delivers DBEvents while executing queries. DBEvents
    # won't be delivered while the application is idle or not using
    # the DBMS session. Running dbevent_getter() ensures a DBEvent is
    # delivered within the polling interval even when the DBMS session 
    # is idle

    gvp = py.IIAPI_GETEVENTPARM()
    gvp.gv_connHandle = connHandle
    gvp.gv_timeout = 0
    while run_flag.is_set():
        async with session_mutex:
            await py.IIapi_getEvent(gvp)
        # repeat after half a polling interval
        await asyncio.sleep(DBEVENT_POLLING_INTERVAL * 0.5)
...