Skip to content

Navigation Menu

Sign in
Appearance settings

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

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit eda2d30

Browse files
committed
Fix linting errors and improve located error test
1 parent 17be672 commit eda2d30

File tree

5 files changed

+321
-295
lines changed

5 files changed

+321
-295
lines changed

‎src/graphql/error/located_error.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@
55
from contextlib import suppress
66
from typing import TYPE_CHECKING, Collection
77

8+
from ..language.source import Source, is_source
89
from ..pyutils import inspect
9-
from ..language.source import is_source, Source
1010
from .graphql_error import GraphQLError
1111

1212
if TYPE_CHECKING:
@@ -40,7 +40,7 @@ def located_error(
4040
except AttributeError:
4141
message = str(original_error)
4242
try:
43-
source = original_error.source
43+
source = original_error.source# type: ignore
4444
if not is_source(source):
4545
source = Source(source) if isinstance(source, str) else None
4646
except AttributeError:
Lines changed: 135 additions & 135 deletions
Original file line numberDiff line numberDiff line change
@@ -1,135 +1,135 @@
1-
"""Build field plan"""
2-
3-
from __future__ import annotations
4-
5-
import sys
6-
from typing import TYPE_CHECKING, Dict, NamedTuple
7-
8-
from ..pyutils import RefMap, RefSet
9-
from .collect_fields import DeferUsage, FieldDetails
10-
11-
if TYPE_CHECKING:
12-
from ..language import FieldNode
13-
14-
try:
15-
from typing import TypeAlias
16-
except ImportError: # Python < 3.10
17-
from typing_extensions import TypeAlias
18-
19-
__all__ = [
20-
"DeferUsageSet",
21-
"FieldGroup",
22-
"FieldPlan",
23-
"GroupedFieldSet",
24-
"NewGroupedFieldSetDetails",
25-
"build_field_plan",
26-
]
27-
28-
29-
DeferUsageSet: TypeAlias = RefSet[DeferUsage]
30-
31-
32-
class FieldGroup(NamedTuple):
33-
"""A group of fields with defer usages."""
34-
35-
fields: list[FieldDetails]
36-
defer_usages: DeferUsageSet | None = None
37-
38-
def to_nodes(self) -> list[FieldNode]:
39-
"""Return the field nodes in this group."""
40-
return [field_details.node for field_details in self.fields]
41-
42-
43-
if sys.version_info < (3, 9):
44-
GroupedFieldSet: TypeAlias = Dict[str, FieldGroup]
45-
else: # Python >= 3.9
46-
GroupedFieldSet: TypeAlias = dict[str, FieldGroup]
47-
48-
49-
class NewGroupedFieldSetDetails(NamedTuple):
50-
"""Details of a new grouped field set."""
51-
52-
grouped_field_set: GroupedFieldSet
53-
should_initiate_defer: bool
54-
55-
56-
class FieldPlan(NamedTuple):
57-
"""A plan for executing fields."""
58-
59-
grouped_field_set: GroupedFieldSet
60-
new_grouped_field_set_details_map: RefMap[DeferUsageSet, NewGroupedFieldSetDetails]
61-
62-
63-
def build_field_plan(
64-
fields: dict[str, list[FieldDetails]],
65-
parent_defer_usages: DeferUsageSet | None = None,
66-
) -> FieldPlan:
67-
"""Build a plan for executing fields."""
68-
if parent_defer_usages is None:
69-
parent_defer_usages = RefSet()
70-
71-
grouped_field_set: GroupedFieldSet = {}
72-
73-
new_grouped_field_set_details_map: RefMap[
74-
DeferUsageSet, NewGroupedFieldSetDetails
75-
] = RefMap()
76-
77-
map_: dict[str, tuple[DeferUsageSet, list[FieldDetails]]] = {}
78-
79-
for response_key, field_details_list in fields.items():
80-
defer_usage_set: RefSet[DeferUsage] = RefSet()
81-
in_original_result = False
82-
for field_details in field_details_list:
83-
defer_usage = field_details.defer_usage
84-
if defer_usage is None:
85-
in_original_result = True
86-
continue
87-
defer_usage_set.add(defer_usage)
88-
if in_original_result:
89-
defer_usage_set.clear()
90-
else:
91-
defer_usage_set -= {
92-
defer_usage
93-
for defer_usage in defer_usage_set
94-
if any(
95-
ancestor in defer_usage_set for ancestor in defer_usage.ancestors
96-
)
97-
}
98-
map_[response_key] = (defer_usage_set, field_details_list)
99-
100-
for response_key, [defer_usage_set, field_details_list] in map_.items():
101-
if defer_usage_set == parent_defer_usages:
102-
field_group = grouped_field_set.get(response_key)
103-
if field_group is None: # pragma: no cover else
104-
field_group = FieldGroup([], defer_usage_set)
105-
grouped_field_set[response_key] = field_group
106-
field_group.fields.extend(field_details_list)
107-
continue
108-
109-
for (
110-
new_grouped_field_set_defer_usage_set,
111-
new_grouped_field_set_details,
112-
) in new_grouped_field_set_details_map.items():
113-
if new_grouped_field_set_defer_usage_set == defer_usage_set:
114-
new_grouped_field_set = new_grouped_field_set_details.grouped_field_set
115-
break
116-
else:
117-
new_grouped_field_set = {}
118-
new_grouped_field_set_details = NewGroupedFieldSetDetails(
119-
new_grouped_field_set,
120-
any(
121-
defer_usage not in parent_defer_usages
122-
for defer_usage in defer_usage_set
123-
),
124-
)
125-
new_grouped_field_set_details_map[defer_usage_set] = (
126-
new_grouped_field_set_details
127-
)
128-
129-
field_group = new_grouped_field_set.get(response_key)
130-
if field_group is None: # pragma: no cover else
131-
field_group = FieldGroup([], defer_usage_set)
132-
new_grouped_field_set[response_key] = field_group
133-
field_group.fields.extend(field_details_list)
134-
135-
return FieldPlan(grouped_field_set, new_grouped_field_set_details_map)
1+
"""Build field plan"""
2+
3+
from __future__ import annotations
4+
5+
import sys
6+
from typing import TYPE_CHECKING, Dict, NamedTuple
7+
8+
from ..pyutils import RefMap, RefSet
9+
from .collect_fields import DeferUsage, FieldDetails
10+
11+
if TYPE_CHECKING:
12+
from ..language import FieldNode
13+
14+
try:
15+
from typing import TypeAlias
16+
except ImportError: # Python < 3.10
17+
from typing_extensions import TypeAlias
18+
19+
__all__ = [
20+
"DeferUsageSet",
21+
"FieldGroup",
22+
"FieldPlan",
23+
"GroupedFieldSet",
24+
"NewGroupedFieldSetDetails",
25+
"build_field_plan",
26+
]
27+
28+
29+
DeferUsageSet: TypeAlias = RefSet[DeferUsage]
30+
31+
32+
class FieldGroup(NamedTuple):
33+
"""A group of fields with defer usages."""
34+
35+
fields: list[FieldDetails]
36+
defer_usages: DeferUsageSet | None = None
37+
38+
def to_nodes(self) -> list[FieldNode]:
39+
"""Return the field nodes in this group."""
40+
return [field_details.node for field_details in self.fields]
41+
42+
43+
if sys.version_info < (3, 9):
44+
GroupedFieldSet: TypeAlias = Dict[str, FieldGroup]
45+
else: # Python >= 3.9
46+
GroupedFieldSet: TypeAlias = dict[str, FieldGroup]
47+
48+
49+
class NewGroupedFieldSetDetails(NamedTuple):
50+
"""Details of a new grouped field set."""
51+
52+
grouped_field_set: GroupedFieldSet
53+
should_initiate_defer: bool
54+
55+
56+
class FieldPlan(NamedTuple):
57+
"""A plan for executing fields."""
58+
59+
grouped_field_set: GroupedFieldSet
60+
new_grouped_field_set_details_map: RefMap[DeferUsageSet, NewGroupedFieldSetDetails]
61+
62+
63+
def build_field_plan(
64+
fields: dict[str, list[FieldDetails]],
65+
parent_defer_usages: DeferUsageSet | None = None,
66+
) -> FieldPlan:
67+
"""Build a plan for executing fields."""
68+
if parent_defer_usages is None:
69+
parent_defer_usages = RefSet()
70+
71+
grouped_field_set: GroupedFieldSet = {}
72+
73+
new_grouped_field_set_details_map: RefMap[
74+
DeferUsageSet, NewGroupedFieldSetDetails
75+
] = RefMap()
76+
77+
map_: dict[str, tuple[DeferUsageSet, list[FieldDetails]]] = {}
78+
79+
for response_key, field_details_list in fields.items():
80+
defer_usage_set: RefSet[DeferUsage] = RefSet()
81+
in_original_result = False
82+
for field_details in field_details_list:
83+
defer_usage = field_details.defer_usage
84+
if defer_usage is None:
85+
in_original_result = True
86+
continue
87+
defer_usage_set.add(defer_usage)
88+
if in_original_result:
89+
defer_usage_set.clear()
90+
else:
91+
defer_usage_set -= {
92+
defer_usage
93+
for defer_usage in defer_usage_set
94+
if any(
95+
ancestor in defer_usage_set for ancestor in defer_usage.ancestors
96+
)
97+
}
98+
map_[response_key] = (defer_usage_set, field_details_list)
99+
100+
for response_key, [defer_usage_set, field_details_list] in map_.items():
101+
if defer_usage_set == parent_defer_usages:
102+
field_group = grouped_field_set.get(response_key)
103+
if field_group is None: # pragma: no cover else
104+
field_group = FieldGroup([], defer_usage_set)
105+
grouped_field_set[response_key] = field_group
106+
field_group.fields.extend(field_details_list)
107+
continue
108+
109+
for (
110+
new_grouped_field_set_defer_usage_set,
111+
new_grouped_field_set_details,
112+
) in new_grouped_field_set_details_map.items():
113+
if new_grouped_field_set_defer_usage_set == defer_usage_set:
114+
new_grouped_field_set = new_grouped_field_set_details.grouped_field_set
115+
break
116+
else:
117+
new_grouped_field_set = {}
118+
new_grouped_field_set_details = NewGroupedFieldSetDetails(
119+
new_grouped_field_set,
120+
any(
121+
defer_usage not in parent_defer_usages
122+
for defer_usage in defer_usage_set
123+
),
124+
)
125+
new_grouped_field_set_details_map[defer_usage_set] = (
126+
new_grouped_field_set_details
127+
)
128+
129+
field_group = new_grouped_field_set.get(response_key)
130+
if field_group is None: # pragma: no cover else
131+
field_group = FieldGroup([], defer_usage_set)
132+
new_grouped_field_set[response_key] = field_group
133+
field_group.fields.extend(field_details_list)
134+
135+
return FieldPlan(grouped_field_set, new_grouped_field_set_details_map)

‎src/graphql/utilities/get_introspection_query.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@
1111
try:
1212
from typing import Literal, TypedDict
1313
except ImportError: # Python < 3.8
14-
from typing_extensions import Literal, TypedDict# type: ignore
14+
from typing_extensions import Literal, TypedDict
1515
try:
1616
from typing import TypeAlias
1717
except ImportError: # Python < 3.10

‎tests/error/test_located_error.py‎

Lines changed: 32 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
from typing import Any, cast
22

33
from graphql.error import GraphQLError, located_error
4+
from graphql.language import Source
45

56

67
def describe_located_error():
@@ -36,10 +37,35 @@ def __init__(self):
3637

3738
assert str(located_error(LazyError())) == "lazy"
3839

40+
def handles_error_with_proper_source():
41+
class CustomError(Exception):
42+
source = Source("foo")
43+
44+
e = located_error(CustomError())
45+
46+
assert e.source
47+
assert isinstance(e.source, Source)
48+
assert e.source.body == "foo"
49+
3950
def handles_error_with_str_source():
40-
class CustomException(Exception):
41-
def __init__(self, message, source):
42-
super().__init__(message)
43-
self.source = source
44-
e = located_error(CustomException("msg", "source"))
45-
assert e.source and e.source.get_location(0)
51+
class CustomError(Exception):
52+
source = "foo"
53+
54+
e = located_error(CustomError())
55+
56+
assert e.source
57+
assert isinstance(e.source, Source)
58+
assert e.source.body == "foo"
59+
60+
def handles_error_with_non_source():
61+
class CustomError(Exception):
62+
source = Exception("Not a source")
63+
64+
e = located_error(CustomError())
65+
66+
assert e.source is None
67+
68+
def handles_error_without_source():
69+
e = located_error(ValueError("No source"))
70+
71+
assert e.source is None

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /