WebsocketsTransport¶
The websockets transport supports both:
It will propose both subprotocols to the backend and detect the supported protocol from the response http headers returned by the backend.
Note
For some backends (graphql-ws before version 5.6.1 without backwards compatibility), it may be necessary to specify
only one subprotocol to the backend. It can be done by using
subprotocols=[WebsocketsTransport.GRAPHQLWS_SUBPROTOCOL]
or subprotocols=[WebsocketsTransport.APOLLO_SUBPROTOCOL]
in the transport arguments.
This transport allows to do multiple queries, mutations and subscriptions on the same websocket connection.
Reference: gql.transport.websockets.WebsocketsTransport
import asyncio
import logging
from gql import Client, gql
from gql.transport.websockets import WebsocketsTransport
logging.basicConfig(level=logging.INFO)
async def main():
transport = WebsocketsTransport(url="wss://countries.trevorblades.com/graphql")
# Using `async with` on the client will start a connection on the transport
# and provide a `session` variable to execute queries on this connection
async with Client(
transport=transport,
fetch_schema_from_transport=True,
) as session:
# Execute single query
query = gql(
"""
query getContinents {
continents {
code
name
}
}
"""
)
result = await session.execute(query)
print(result)
# Request subscription
subscription = gql(
"""
subscription {
somethingChanged {
id
}
}
"""
)
async for result in session.subscribe(subscription):
print(result)
asyncio.run(main())
Websockets SSL¶
If you need to connect to an ssl encrypted endpoint:
use
wss
instead ofws
in the url of the transport
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
headers={'Authorization': 'token'}
)
If you have a self-signed ssl certificate, you need to provide an ssl_context with the server public certificate:
import pathlib
import ssl
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
localhost_pem = pathlib.Path(__file__).with_name("YOUR_SERVER_PUBLIC_CERTIFICATE.pem")
ssl_context.load_verify_locations(localhost_pem)
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
ssl=ssl_context
)
If you have also need to have a client ssl certificate, add:
ssl_context.load_cert_chain(certfile='YOUR_CLIENT_CERTIFICATE.pem', keyfile='YOUR_CLIENT_CERTIFICATE_KEY.key')
Websockets authentication¶
There are two ways to send authentication tokens with websockets depending on the server configuration.
Using HTTP Headers
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
headers={'Authorization': 'token'}
)
With a payload in the connection_init websocket message
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
init_payload={'Authorization': 'token'}
)
Keep-Alives¶
Apollo protocol¶
With the Apollo protocol, the backend can optionally send unidirectional keep-alive (“ka”) messages (only from the server to the client).
It is possible to configure the transport to close if we don’t receive a “ka” message
within a specified time using the keep_alive_timeout
parameter.
Here is an example with 60 seconds:
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
keep_alive_timeout=60,
)
One disadvantage of the Apollo protocol is that because the keep-alives are only sent from the server to the client, it can be difficult to detect the loss of a connection quickly from the server side.
GraphQL-ws protocol¶
With the GraphQL-ws protocol, it is possible to send bidirectional ping/pong messages. Pings can be sent either from the client or the server and the other party should answer with a pong.
As with the Apollo protocol, it is possible to configure the transport to close if we don’t
receive any message from the backend within the specified time using the keep_alive_timeout
parameter.
But there is also the possibility for the client to send pings at a regular interval and verify
that the backend sends a pong within a specified delay.
This can be done using the ping_interval
and pong_timeout
parameters.
Here is an example with a ping sent every 60 seconds, expecting a pong within 10 seconds:
transport = WebsocketsTransport(
url='wss://SERVER_URL:SERVER_PORT/graphql',
ping_interval=60,
pong_timeout=10,
)