On previous versions of GQL, the code was sync only , it means that when you ran execute on the Client, you could do nothing else in the current Thread and had to wait for an answer or a timeout from the backend to continue. The only http library was requests, allowing only sync usage.
From the version 3 of GQL, we support sync and async transports using asyncio.
With the async transports, there is now the possibility to execute GraphQL requests asynchronously, allowing to execute multiple requests in parallel if needed.
If you don’t care or need async functionality, it is still possible, with async transports, to run the execute or subscribe methods directly from the Client (as described in the Sync Usage example) and GQL will execute the request in a synchronous manner by running an asyncio event loop itself.
This won’t work though if you already have an asyncio event loop running. In that case you should use the async methods.
Example¶If you use an async transport, you can use GQL asynchronously using asyncio.
put your code in an asyncio coroutine (method starting with async def)
use async with client as session: to connect to the backend and provide a session instance
use the await keyword to execute requests: await session.execute(...)
then run your coroutine in an asyncio event loop by running asyncio.run
import asyncio
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
async def main():
# Select your transport with a defined url endpoint
transport = AIOHTTPTransport(url="https://countries.trevorblades.com/graphql")
# Create a GraphQL client using the defined transport
client = Client(transport=transport)
# Provide a GraphQL query
query = gql(
"""
query getContinents {
continents {
code
name
}
}
"""
)
# 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 as session:
# Execute the query
result = await session.execute(query)
print(result)
asyncio.run(main())
IPython¶
Warning
On some Python environments, like Jupyter or Spyder, which are using IPython, an asyncio event loop is already created for you by the environment.
In this case, running the above code might generate the following error:
RuntimeError: asyncio.run() cannot be called from a running event loop
If that happens, depending on the environment, you should replace asyncio.run(main()) by either:
OR:
loop = asyncio.get_running_loop() loop.create_task(main())
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4