Are websockets supported?

[ecthiender]

GraphQL spec says it is transport agnostic, but most implementations support HTTP and the apollo protocol over websockets.

Looking here: https://github.com/graphql-python/gql/tree/master/gql/transport it does not look like websockets are supported. But I may be wrong.

I need websocket support for my project, looking for a Python client that has support for the same.

[waterdrop01]

Websocket support would indeed be great, so we can have subscriptions :)

[remorses]

I can’t find any library for graphql subscriptions in python, we need to implement this asap

[loic001]

FYI, another good GraphQL python client here. Websocket not yet implemented but they plan to do so.

[ecthiender]

I just published a python GraphQL client library supporting websockets. As I immediately needed it for my project, there is support for only websocket transport as of now. Check it out here. Contributions and comments are most welcome.

pip package
Source

[leszekhanusz]

I created the pull request #70 to have a websockets transport to connect to apollo servers
It needs Python 3.6 and asyncio (so tests are failing ...)

Supports fetching schema from transport using the introspection query
Supports queries, mutations AND subscriptions
Supports connection using a client ssl certificate

Only one request per connection for now

[leszekhanusz]

Now it is working with multiple requests per connection.
Supports multiple subscriptions in parallel.

[Cito]

@ecthiender This is now available in v3.0.0a0 (see README). Please let us know how it is working for you.

[ecthiender]

@Cito umm.. I needed this functionality back when I was looking for it. So I implemented my own library: https://github.com/ecthiender/py-graphql-client

Thanks!

[jameslewis4891]

Hello @leszekhanusz regarding #42 (comment)

Do you have an example of this please? No matter what I try with asyncio making use of the example code in https://github.com/graphql-python/gql/blob/master/docs/code_examples/websockets_async.py (for subscriptions only) my app always blocks. I cannot figure out a way of running my gql subscriptions in parallel aside form the main thread (thats running Flask)

[leszekhanusz]

@jameslewis4891 There is an example in the gql docs, basically you have to create multiple asyncio tasks. Please note that it is using asyncio so it is always using the same Thread.

[jameslewis4891]

@jameslewis4891 There is an example in the gql docs, basically you have to create multiple asyncio tasks. Please note that it is using asyncio so it is always using the same Thread.

Thanks @leszekhanusz that works great however I am struggling to test this code I am trying to mock the async with client as session: line and running into difficulties

My Code

class GraphQLSubscriber:

    def get_subscribing_to(self) -> str:
        pass

    def getName(self) -> str:
        pass

    def getQuery(self) -> str:
        pass

    def getHandler(self) -> Callable:
        pass

    def getVariables(self) -> dict:
        pass


def connection_backoff(data: dict):
    log_message('Connection error to GraphQL - Reattempting connection', error=True)


class GraphQLSubscriptionHandler(threading.Thread):

    def __init__(self, config, subscribers):
        super(GraphQLSubscriptionHandler, self).__init__(daemon=True)
        self.subscribers = subscribers
        self.graphql_url = get_config_value(API_URL, config)

    async def execute_subscription(self, session, document, variable_values, handler):
        async for result in session.subscribe(document=document, variable_values=variable_values):
            handler(result)

    @backoff.on_exception(backoff.expo, Exception, max_time=1000, on_backoff=connection_backoff)
    async def init_subscriptions(self):
        transport = WebsocketsTransport(url=self.graphql_url)
    
        log_message(f'Connecting to graphql server [{self.graphql_url}')
    
        client = Client(transport=transport, fetch_schema_from_transport=True)
    
        async with client as session:
            tasks = []
    
            for sub in self.subscribers:
                log_message(f'GQL Subscriber [{sub.getName()}] Subscribing to [{sub.get_subscribing_to()}]')
                tasks.append(asyncio.create_task(self.execute_subscription(session, document=gql(sub.getQuery()),
                                                                           variable_values=sub.getVariables(),
                                                                           handler=sub.getHandler())))
            await asyncio.gather(*tasks)

    def run(self):
        asyncio.run(self.init_subscriptions())

My test

CONFIG = {
    "API_URL": "ws://localhost",
}

@pytest.mark.asyncio
async def test_should_subscribe(event_loop):
    mock_subscriber = Mock()

    with patch('app.graphql.graphql_subscription_handler.WebsocketsTransport') as gql_websockets_transport_mock:
        with patch('app.graphql.graphql_subscription_handler.Client') as gql_client_mock:

            # arrange
            mock_session = Mock()

            gql_client_instance = gql_client_mock.return_value

            gql_client_instance.blah.return_value = mock_session

            gql_client_instance.__aenter__.return_value = mock_session
            gql_client_instance.side_effect = mock_session

            graphql_subscription_handler = GraphQLSubscriptionHandler(CONFIG, [mock_subscriber])

            # act
            await graphql_subscription_handler.init_subscriptions()

            # assert
            mock_subscriber.getQuery.assert_called_once()

I keep getting Attribute errors on mocking the aenter routine. I am new to coroutines in general and particularly in Python

I am interested in testing that the session has been called with the correct variables from the passed in subscriber