How do I communicate using gql with Apollo websocket protocol

Question

I want to connect to the websocket. When I inspect the traffic between the client and the server I see that the first message is the handshake:

{"type": "connection_init",
 "payload":
           {
            "accept-language": "en",
            "ec-version": "5.1.88",
            "referrer": URL,
           }
}

Based on the format (the keys of the dict) I conclude that the websocket uses an Apollo websocket transport prototcol.

Next, I am following the websocket-example of gql's documentation.

import asyncio
import logging
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport


logging.basicConfig(level=logging.INFO)
    
async def main():
    transport = WebsocketsTransport(url=URL,
                                    init_payload={'accept-language': 'en',
                                                  'ec-version': '5.1.88'})
    async with Client(transport=transport,
                      fetch_schema_from_transport=False) as session:
        # do something
    

asyncio.run(main())

After reading more about the protocol here I still don't understand how I can send messages to the server within my Python-script How do I send the the below message to the websocket?

{
 "id": "1073897396",
 "type": "start",
 "operationName": operation,
 "eventId": 488
}

Answer 1

The EuropeanTour website does not use the Apollo websocket protocol for its API so you cannot use the standard WebsocketTransport class.

What I found out:

• instead of returning a connection_ack message after the connection_init, it returns a wsIdentity message
• for a subscription, the client will first send a start message containing an operationName and an eventId, without any GraphQL query
• the server will then answer with a request-full-subscription message, requiring the client to send the full subscription query
• the client has then to send a full-subscription message containing the full requested query, including the operationName and a new field subscriptionName
• the id is not actually random, it is generated from hashing together the subscriptionName and the variables, from a json format with the dictionary keys sorted. It took me a while to figure it out as no error is generated if the id is not correct
• the server seems to cache the GraphQL subscription based on the IP address probably so it will not request the subscription with a request-full-subscription message if it has already received it previously

Here is a Chrome screenshot showing the request-full-subscription messages:

You can then create your own transport by inheriting the WebsocketTransport class and making the necessary modifications.

Here is an example of working code:

import asyncio
import json
import logging
from typing import Any, AsyncGenerator, Dict, Optional, Tuple

from graphql import DocumentNode, ExecutionResult, print_ast

from gql import Client, gql
from gql.transport.exceptions import TransportProtocolError
from gql.transport.websockets import WebsocketsTransport

logging.basicConfig(level=logging.INFO)


class EuropeanTourWebsocketsTransport(WebsocketsTransport):
    def _hash(self, e):
        t = 5381
        r = len(e)
        while r:
            r -= 1
            t = t * 33 ^ ord(e[r])
        return t & 0xFFFFFFFF

    def _calc_id(self, subscription_name, variables):

        obj = {
            "subscriptionName": subscription_name,
            "variables": variables,
        }

        obj_stringified = json.dumps(
            obj,
            separators=(",", ":"),
            sort_keys=True,
        )

        hashed_value = self._hash(obj_stringified)

        return hashed_value

    async def _send_query(
        self,
        document: DocumentNode,
        variable_values: Optional[Dict[str, Any]] = None,
        operation_name: Optional[str] = None,
    ) -> int:

        # Calculate the id by hashing the subscription name and the variables
        query_id = self._calc_id(self.latest_subscription_name, variable_values)

        # Creating the payload for the full subscription
        payload: Dict[str, Any] = {"query": print_ast(document)}
        if variable_values:
            payload["variables"] = variable_values
        if operation_name:
            payload["operationName"] = operation_name
            payload["subscriptionName"] = self.latest_subscription_name

        # Saving the full query first and waiting for the server to request it later
        self.saved_full_subscriptions[str(query_id)] = payload

        # Then first start to request the subscription only with the operation name
        query_str = json.dumps(
            {
                "id": str(query_id),
                "type": "start",
                "operationName": operation_name,
                "eventId": self.latest_event_id,
            }
        )

        await self._send(query_str)

        return query_id

    async def subscribe(
        self,
        document: DocumentNode,
        *,
        variable_values: Optional[Dict[str, Any]] = None,
        operation_name: str,
        subscription_name: str,
        event_id: int,
        send_stop: Optional[bool] = True,
    ) -> AsyncGenerator[ExecutionResult, None]:

        self.latest_event_id = event_id
        self.latest_subscription_name = subscription_name

        async for result in super().subscribe(
            document,
            variable_values=variable_values,
            operation_name=operation_name,
            send_stop=send_stop,
        ):
            yield result

    async def _wait_ack(self) -> None:

        self.saved_full_subscriptions = {}

        while True:
            init_answer = await self._receive()

            answer_type, answer_id, execution_result = self._parse_answer(init_answer)

            if answer_type == "wsIdentity":
                return

            raise TransportProtocolError(
                "Websocket server did not return a wsIdentity response"
            )

    def _parse_answer(
        self, answer: str
    ) -> Tuple[str, Optional[int], Optional[ExecutionResult]]:
        try:
            json_answer = json.loads(answer)
        except ValueError:
            raise TransportProtocolError(
                f"Server did not return a GraphQL result: {answer}"
            )

        if "wsIdentity" in json_answer:
            return ("wsIdentity", json_answer["wsIdentity"], None)

        elif (
            "type" in json_answer and json_answer["type"] == "request-full-subscription"
        ):
            return ("request-full-subscription", json_answer["id"], None)

        else:

            return self._parse_answer_apollo(json_answer)

    async def send_full_subscription(self, answer_id: str):

        if answer_id not in self.saved_full_subscriptions:
            raise Exception(f"Full subscription not found for id {answer_id}")

        payload = self.saved_full_subscriptions[answer_id]

        query_str = json.dumps(
            {"id": answer_id, "type": "full-subscription", "payload": payload}
        )

        await self._send(query_str)

    async def _handle_answer(
        self,
        answer_type: str,
        answer_id: Optional[int],
        execution_result: Optional[ExecutionResult],
    ) -> None:

        if answer_type == "request-full-subscription":
            await self.send_full_subscription(answer_id)

        else:
            await super()._handle_answer(answer_type, answer_id, execution_result)


async def main():

    transport = EuropeanTourWebsocketsTransport(
        url="wss://btec-websocket.services.imgarena.com",
        init_payload={
            "accept-language": "en",
            "ec-version": "5.1.88",
            "operator": "europeantour",
            "referrer": "https://www.europeantour.com/",
            "sport": "GOLF",
        },
    )

    async with Client(
        transport=transport, fetch_schema_from_transport=False
    ) as session:

        query = gql(
            """
subscription ShotTrackerSubscribeToGolfTournamentGroupScores($input: SubscribeToGolfTournamentGroupScoresInput!) {
  subscribeToGolfTournamentGroupScores(input: $input) {
    groupId
    l1Course
    teamId
    players {
      id
      lastName
      firstName
    }
    roundScores {
      courseId
      roundNo
      toParToday {
        value
      }
      holesThrough {
        value
      }
      startHole
      holes {
        holePar
        holeStrokes
        holeOrder
        holeNumber
      }
      isPlayoff
    }
    toPar {
      value
    }
    tournamentPosition {
      format
      value
      displayValue
    }
    status
  }
}
"""
        )

        variables = {
            "input": {
                "teamId": 21,
                "tournamentId": 488,
                "roundNo": 4,
            },
        }

        async for result in session.subscribe(
            query,
            operation_name="ShotTrackerSubscribeToGolfTournamentGroupScores",
            variable_values=variables,
            subscription_name="subscribeToGolfTournamentGroupScores",
            event_id=488,
        ):

            print(result)


asyncio.run(main())

Answer 2

Considering the JSON message you provided:

{
    "id": "1073897396",
    "type": "start",
    "operationName": "SubscribeToGolfTournamentStatus",
    "eventId": 488
}

The operationName suggests that you are trying to start a subscription named SubscribeToGolfTournamentStatus. You would need the actual GraphQL subscription query that this name refers to.

So in your case, to send a message to the WebSocket, you will typically need to execute a GraphQL subscription. In the context of the Apollo WebSocket protocol and using the gql library, you will use the session.subscribe() method.

Let's assume the subscription query looks like this (this is just a placeholder, you would need the real subscription query structure):

subscription SubscribeToGolfTournamentStatus($eventId: Int!) {
    golfTournamentStatus(eventId: $eventId) {
        ...YourDesiredFieldsHere
    }
}

Here is how you would send that subscription using gql:

import asyncio
import logging
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport

logging.basicConfig(level=logging.INFO)

async def main():
    transport = WebsocketsTransport(url='wss://btec-websocket.services.imgarena.com',
                                    init_payload={
                                        'accept-language': 'en',
                                        'ec-version': '5.1.88',
                                        'operator': 'europeantour',
                                        'referrer': 'https://www.europeantour.com/',
                                        'sport': 'GOLF'
                                    })
    async with Client(transport=transport, fetch_schema_from_transport=False) as session:
        
        # Define the subscription
        subscription = gql('''
        subscription SubscribeToGolfTournamentStatus($eventId: Int!) {
            golfTournamentStatus(eventId: $eventId) {
                ...YourDesiredFieldsHere
            }
        }
        ''')

        # Subscribe and handle incoming data
        async for result in session.subscribe(subscription, variable_values={"eventId": 488}):
            print(result)

asyncio.run(main())

That would:

• define the GraphQL subscription with the gql function (Remember, you will need to replace the placeholder GraphQL query with the actual structure you are targeting).
  You can see an example in leszekhanusz/gql-actioncable.
• use the session.subscribe() method to start the subscription and pass the necessary variables (eventId in this case).
• set up an asynchronous loop to handle incoming data from the subscription.

Compared to the official example:

• in both case cases, you initialize the WebsocketsTransport with a given URL and payload, and then create a Client session.
• the "do something" part of the example is expanded to specifically show how you would execute a subscription query with the gql library.