Commit 57c083d

authored

Cancel remaining fields on exceptions (#238)

1 parent 0107e30 commit 57c083dCopy full SHA for 57c083d

File tree

2 files changed

+73

-7

lines changed

src/graphql/execution
- execute.py
tests/execution
- test_parallel.py

2 files changed

+73

-7

lines changed

`‎src/graphql/execution/execute.py‎`

Lines changed: 33 additions & 7 deletions

Original file line number	Diff line number	Diff line change
`@@ -2,7 +2,14 @@`
`2`	`2`
`3`	`3`	`from __future__ import annotations`
`4`	`4`
`5`		`-from asyncio import ensure_future, gather, shield, wait_for`
	`5`	`+from asyncio import (`
	`6`	`+ CancelledError,`
	`7`	`+ create_task,`
	`8`	`+ ensure_future,`
	`9`	`+ gather,`
	`10`	`+ shield,`
	`11`	`+ wait_for,`
	`12`	`+)`
`6`	`13`	`from contextlib import suppress`
`7`	`14`	`from copy import copy`
`8`	`15`	`from typing import (`
`@@ -459,12 +466,23 @@ async def get_results() -> dict[str, Any]:`
`459`	`466`	`field = awaitable_fields[0]`
`460`	`467`	`results[field] = await results[field]`
`461`	`468`	`else:`
`462`		`- results.update(`
`463`		`- zip(`
`464`		`- awaitable_fields,`
`465`		`- await gather(*(results[field] for field in awaitable_fields)),`
`466`		`- )`
`467`		`- )`
	`469`	`+ tasks = {`
	`470`	`+ create_task(results[field]): field # type: ignore[arg-type]`
	`471`	`+ for field in awaitable_fields`
	`472`	`+ }`
	`473`	`+`
	`474`	`+ try:`
	`475`	`+ awaited_results = await gather(*tasks)`
	`476`	`+ except Exception:`
	`477`	`+ # Cancel unfinished tasks before raising the exception`
	`478`	`+ for task in tasks:`
	`479`	`+ if not task.done():`
	`480`	`+ task.cancel()`
	`481`	`+ await gather(*tasks, return_exceptions=True)`
	`482`	`+ raise`
	`483`	`+`
	`484`	`+ results.update(zip(awaitable_fields, awaited_results))`
	`485`	`+`
`468`	`486`	`return results`
`469`	`487`
`470`	`488`	`return get_results()`
`@@ -538,6 +556,10 @@ async def await_completed() -> Any:`
`538`	`556`	`try:`
`539`	`557`	`return await completed`
`540`	`558`	`except Exception as raw_error:`
	`559`	`+ # Before Python 3.8 CancelledError inherits Exception and`
	`560`	`+ # so gets caught here.`
	`561`	`+ if isinstance(raw_error, CancelledError):`
	`562`	`+ raise # pragma: no cover`
`541`	`563`	`self.handle_field_error(`
`542`	`564`	`raw_error,`
`543`	`565`	`return_type,`
`@@ -745,6 +767,10 @@ async def complete_awaitable_value(`
`745`	`767`	`if self.is_awaitable(completed):`
`746`	`768`	`completed = await completed`
`747`	`769`	`except Exception as raw_error:`
	`770`	`+ # Before Python 3.8 CancelledError inherits Exception and`
	`771`	`+ # so gets caught here.`
	`772`	`+ if isinstance(raw_error, CancelledError):`
	`773`	`+ raise # pragma: no cover`
`748`	`774`	`self.handle_field_error(`
`749`	`775`	`raw_error, return_type, field_group, path, incremental_data_record`
`750`	`776`	`)`

`‎tests/execution/test_parallel.py‎`

Lines changed: 40 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,7 @@`
`11`	`11`	`GraphQLInt,`
`12`	`12`	`GraphQLInterfaceType,`
`13`	`13`	`GraphQLList,`
	`14`	`+ GraphQLNonNull,`
`14`	`15`	`GraphQLObjectType,`
`15`	`16`	`GraphQLSchema,`
`16`	`17`	`GraphQLString,`
`@@ -193,3 +194,42 @@ async def is_type_of_baz(obj, *_args):`
`193`	`194`	`{"foo": [{"foo": "bar", "foobar": 1}, {"foo": "baz", "foobaz": 2}]},`
`194`	`195`	`None,`
`195`	`196`	`)`
	`197`	`+`
	`198`	`+ @pytest.mark.asyncio`
	`199`	`+ async def cancel_on_exception():`
	`200`	`+ barrier = Barrier(2)`
	`201`	`+ completed = False`
	`202`	`+`
	`203`	`+ async def succeed(*_args):`
	`204`	`+ nonlocal completed`
	`205`	`+ await barrier.wait()`
	`206`	`+ completed = True # pragma: no cover`
	`207`	`+`
	`208`	`+ async def fail(*_args):`
	`209`	`+ raise RuntimeError("Oops")`
	`210`	`+`
	`211`	`+ schema = GraphQLSchema(`
	`212`	`+ GraphQLObjectType(`
	`213`	`+ "Query",`
	`214`	`+ {`
	`215`	`+ "foo": GraphQLField(GraphQLNonNull(GraphQLBoolean), resolve=fail),`
	`216`	`+ "bar": GraphQLField(GraphQLBoolean, resolve=succeed),`
	`217`	`+ },`
	`218`	`+ )`
	`219`	`+ )`
	`220`	`+`
	`221`	`+ ast = parse("{foo, bar}")`
	`222`	`+`
	`223`	`+ awaitable_result = execute(schema, ast)`
	`224`	`+ assert isinstance(awaitable_result, Awaitable)`
	`225`	`+ result = await asyncio.wait_for(awaitable_result, 1.0)`
	`226`	`+`
	`227`	`+ assert result == (`
	`228`	`+ None,`
	`229`	`+ [{"message": "Oops", "locations": [(1, 2)], "path": ["foo"]}],`
	`230`	`+ )`
	`231`	`+`
	`232`	`+ # Unblock succeed() and check that it does not complete`
	`233`	`+ await barrier.wait()`
	`234`	`+ await asyncio.sleep(0)`
	`235`	`+ assert not completed`

0 commit comments

Comments

(0)

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

Commit 57c083d

File tree

2 files changed

2 files changed

`‎src/graphql/execution/execute.py‎`

`‎tests/execution/test_parallel.py‎`

0 commit comments