-
Notifications
You must be signed in to change notification settings - Fork 183
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
add support for yield from graphql_async(schema, query)
#254
Conversation
cc @syrusakbary |
await graphql_async(schema, query)
yield from graphql_async(schema, query)
tests are failing because |
The inspiration for this feature came from here. import asyncio
from graphql import graphql
from graphql.execution.executors.asyncio import AsyncioExecutor
from graphql.type import GraphQLField, GraphQLObjectType, GraphQLSchema, GraphQLString
async def resolver(context, *_):
await asyncio.sleep(0.001)
return "hey"
async def resolver_2(context, *_):
await asyncio.sleep(0.003)
return "hey2"
def resolver_3(contest, *_):
return "hey3"
Type = GraphQLObjectType(
"Type",
{
"a": GraphQLField(GraphQLString, resolver=resolver),
"b": GraphQLField(GraphQLString, resolver=resolver_2),
"c": GraphQLField(GraphQLString, resolver=resolver_3),
},
)
async def get_result(query):
# AsyncioExecutor will try to execute `loop.run_until_complete`
# But since trinity is already running `run_forever`,
# it will result in a error: `RuntimeError('This event loop is already running',)`
result = graphql(GraphQLSchema(Type), query, executor=AsyncioExecutor())
return result
if __name__ == "__main__":
# think of this loop used by an async framework event loop
loop = asyncio.get_event_loop()
query = "{a, b, c}"
# The async will have its own event_loop running
result = loop.run_until_complete(get_result(query))
print('result', result.data)
print('error', result.errors) This gives the following output:
If I try to use a async def get_result(query):
result = graphql(GraphQLSchema(Type), query, executor=AsyncioExecutor(loop=asyncio.new_event_loop()))
return result The output of this code is:
So the only fix I could think of was supporting |
looks like this PR is not needed. I didn't know about loop = asyncio.get_event_loop()
async def do_exec():
result = await execute(
GraphQLSchema(Type),
ast,
executor=AsyncioExecutor(loop),
return_promise=True,
)
assert not result.errors
assert result.data == {"a": "hey", "b": "hey2", "c": "hey3"}
loop.run_until_complete(do_exec()) |
What was wrong?
graphql(schema, query, executor=AsyncioExecutor())
is blocking.grapqhl
is executed with anotherasync
framework which already has an event loop running. The following error is raised in such cases:How was it fixed?
await graphql_async(schema, query)
syntaxAsyncioExecutor
. A new coroutinewait_until_finished_async
has been implemented.wait_until_finished_async
raisesNotImplementedError
error for all other executors.GraphQLCoreBackend
. This has been done by implementing a newcoroutine
document_from_string_async
.document_from_string_async
raisesNotImplementedError
for other backends although this support can be added if needed.