MMM Events API

import asyncio
import base64
import datetime
from asyncio import CancelledError

from aiohttp import BasicAuth
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.exceptions import TransportQueryError
from gql.transport.websockets import WebsocketsTransport


# https://gql.readthedocs.io/en/latest/usage/index.html

class MyClient:
    def __init__(self, base_url, subscription_id, user_name, password, ack_limit=10):
        self.subscription_id = subscription_id

        http_transport = AIOHTTPTransport(url=f"http://{base_url}/graphql", auth=BasicAuth(user_name, password))
        self.http_client = Client(transport=http_transport, fetch_schema_from_transport=False)

        auth_header = f"Basic {base64.b64encode(f'{user_name}:{password}'.encode('ascii')).decode('ascii')}"
        ws_transport = WebsocketsTransport(url=f"ws://{base_url}/subscriptions", init_payload={
            "authorization": auth_header,
        }, close_timeout=600, ack_timeout=600)

        self.ws_client = Client(transport=ws_transport, fetch_schema_from_transport=False)
        self.ack_limit = ack_limit
        self.event_count_to_ack = 0

    async def subscribe(self, entity_type=None):
        filter_fragment = f' entityType:"{entity_type}",' if entity_type else ""
        # You can specify subscription parameters to get events about specific entities: entityType/Parent/Id
        # You can also specify transport details like ackLimit, chunkSize, chunkMaxDelay
        query_text = f"""subscription ($id: GID!, $ackLimit: Int!) {{
        _externalEvents(subscriptionId: $id, {filter_fragment}ackLimit: $ackLimit, chunkSize:10, chunkMaxDelay:2000, excludeOwnModifications:false){{
            events{{
                id
                timestamp
                entityId
                entityType
                entityStatus
            }}
        }}
    }}"""

        query = gql(query_text)
        try:
            params = {"id": self.subscription_id, "ackLimit": str(self.ack_limit)}

            async for result in self.ws_client.subscribe_async(query, params):
                await self.__process_events(result)
        except CancelledError:
            pass

    async def __process_events(self, result):
        print(datetime.datetime.now(), result)
        events = result["_externalEvents"]["events"]
        self.event_count_to_ack += len(events)
        if self.event_count_to_ack >= self.ack_limit:
            last_id = events[-1]["id"]
            await self.__acknowledge(last_id)
            self.event_count_to_ack = 0

    async def __acknowledge(self, last_id):
        query = gql("""mutation($id: GID!, $lastEventId: Long!){
      _acknowledgeExternalEvents(subscriptionId: $id, lastEventId: $lastEventId)
    }""")
        params = {"id": self.subscription_id, "lastEventId": str(last_id)}

        try:
            result = await self.http_client.execute_async(query, params)
            print(f"ack ({last_id}), result: [{result}]")
        except TransportQueryError as e:
            print(f"ack failed: {e.errors}")

    async def unregister(self):
        query = gql("""mutation($id: GID!){
      _unsubscribeExternalEvents(subscriptionId: $id)
    }""")
        params = {"id": self.subscription_id}

        try:
            result = await self.http_client.execute_async(query, params)
            print(f"unregister result {result}")
        except TransportQueryError as e:
            print(f"unregister failed: {e.errors}")

    async def get_status(self):
        query = gql("""query status($id: GID!){
    _externalEventsSubscriptionStatus(subscriptionId:$id){
        subscriptionId
        undeliveredEventCount
        status
    }
  }""")
        params = {"id": self.subscription_id}

        try:
            result = await self.http_client.execute_async(query, params)
            print(f"get_status, result: [{result}]")
        except TransportQueryError as e:
            print(f"get_status failed: {e.errors}")


async def main():
    client = MyClient("localhost:8021", "ae92c852-43a7-4942-b274-29aa252c205b", "admin", "admin")
    await client.get_status()
    #await client.unregister()
    print("Listening for events")
    await client.subscribe(entity_type=None)


asyncio.run(main())