16081423623

Committed 04 Jul 2025 09:57PM UTC coverage: 97.685%. First build

Build # 16081423623

Build Type

Pull #33

github

Committed by

web-flow

Commit Message

Merge 6eb57167c into 784a68391

Pull Request Pull Request #33: Add Subscriptions

Run Details

1798 of 1841 branches covered (97.66%)

Branch coverage included in aggregate %.

1009 of 1176 new or added lines in 36 files covered. (85.8%)

26853 of 27489 relevant lines covered (97.69%)

8.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.0

/undine/utils/graphql/utils.py

from __future__ import annotations

from contextlib import contextmanager
from typing import TYPE_CHECKING, Any, TypeGuard, TypeVar

from django.db.models import ForeignKey
from graphql import (
    DocumentNode,
    FieldNode,
    GraphQLEnumType,
    GraphQLError,
    GraphQLIncludeDirective,
    GraphQLInputObjectType,
    GraphQLInterfaceType,
    GraphQLObjectType,
    GraphQLScalarType,
    GraphQLSkipDirective,
    GraphQLUnionType,
    OperationDefinitionNode,
    OperationType,
    get_argument_values,
    get_directive_values,
)

from undine.exceptions import (
    DirectiveLocationError,
    GraphQLGetRequestMultipleOperationsNoOperationNameError,
    GraphQLGetRequestNonQueryOperationError,
    GraphQLGetRequestNoOperationError,
    GraphQLGetRequestOperationNotFoundError,
)
from undine.utils.text import to_snake_case

if TYPE_CHECKING:
    from collections.abc import Generator, Iterable

    from graphql import (
        DirectiveLocation,
        DocumentNode,
        GraphQLList,
        GraphQLNonNull,
        GraphQLOutputType,
        GraphQLWrappingType,
        SelectionNode,
    )
    from graphql.execution.values import NodeWithDirective

    from undine import Field, GQLInfo
    from undine.directives import Directive
    from undine.typing import ModelField


__all__ = [
    "check_directives",
    "get_arguments",
    "get_queried_field_name",
    "get_underlying_type",
    "is_connection",
    "is_edge",
    "is_node_interface",
    "is_page_info",
    "is_relation_id",
    "is_subscription_operation",
    "should_skip_node",
    "with_graphql_error_path",
]


TGraphQLType = TypeVar(
    "TGraphQLType",
    GraphQLScalarType,
    GraphQLObjectType,
    GraphQLInterfaceType,
    GraphQLUnionType,
    GraphQLEnumType,
    GraphQLInputObjectType,
)


# Getters


def get_underlying_type(
    gql_type: (
        TGraphQLType
        | GraphQLList[TGraphQLType]
        | GraphQLList[GraphQLNonNull[TGraphQLType]]
        | GraphQLNonNull[TGraphQLType]
        | GraphQLNonNull[GraphQLList[TGraphQLType]]
        | GraphQLNonNull[GraphQLList[GraphQLNonNull[TGraphQLType]]]
        | GraphQLWrappingType[TGraphQLType]
    ),
) -> TGraphQLType:
    while hasattr(gql_type, "of_type"):
        gql_type = gql_type.of_type
    return gql_type


def get_arguments(info: GQLInfo) -> dict[str, Any]:
    """Get input arguments for the current field from the GraphQL resolve info."""
    graphql_field = info.parent_type.fields[info.field_name]
    return get_argument_values(graphql_field, info.field_nodes[0], info.variable_values)


def get_queried_field_name(original_name: str, info: GQLInfo) -> str:
    """Get the name of a field in the current query."""
    return original_name if info.path.key == info.field_name else info.path.key  # type: ignore[return-value]


async def pre_evaluate_request_user(info: GQLInfo) -> None:
    """
    Fetches the request user from the context and caches it to the request.
    This is a workaround when current user is required in an async event loop,
    but the function itself is not async.
    """
    # '_current_user' would be set by 'django.contrib.auth.middleware.get_user' when calling 'request.user'
    info.context._cached_user = await info.context.auser()  # type: ignore[attr-defined]  # noqa: SLF001


# Predicates


def is_connection(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:
    return (
        isinstance(field_type, GraphQLObjectType)
        and field_type.name.endswith("Connection")
        and "pageInfo" in field_type.fields
        and "edges" in field_type.fields
    )


def is_edge(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:
    return (
        isinstance(field_type, GraphQLObjectType)
        and field_type.name.endswith("Edge")
        and "cursor" in field_type.fields
        and "node" in field_type.fields
    )


def is_node_interface(field_type: GraphQLOutputType) -> TypeGuard[GraphQLInterfaceType]:
    return (
        isinstance(field_type, GraphQLInterfaceType)  # comment here for better formatting
        and field_type.name == "Node"
        and "id" in field_type.fields
    )


def is_page_info(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:
    return (
        isinstance(field_type, GraphQLObjectType)
        and field_type.name == "PageInfo"
        and "hasNextPage" in field_type.fields
        and "hasPreviousPage" in field_type.fields
        and "startCursor" in field_type.fields
        and "endCursor" in field_type.fields
    )


def is_typename_metafield(field_node: SelectionNode) -> TypeGuard[FieldNode]:
    if not isinstance(field_node, FieldNode):
        return False
    return field_node.name.value.lower() == "__typename"


def is_relation_id(field: ModelField, field_node: FieldNode) -> TypeGuard[Field]:
    return isinstance(field, ForeignKey) and field.get_attname() == to_snake_case(field_node.name.value)


def is_subscription_operation(document: DocumentNode) -> bool:
    if len(document.definitions) != 1:
        return False

    operation_definition = document.definitions[0]
    if not isinstance(operation_definition, OperationDefinitionNode):
        return False

    return operation_definition.operation == OperationType.SUBSCRIPTION


def should_skip_node(node: NodeWithDirective, variable_values: dict[str, Any]) -> bool:
    skip_args = get_directive_values(GraphQLSkipDirective, node, variable_values)
    if skip_args is not None and skip_args["if"] is True:
        return True

    include_args = get_directive_values(GraphQLIncludeDirective, node, variable_values)
    return include_args is not None and include_args["if"] is False


# Misc.


@contextmanager
def with_graphql_error_path(info: GQLInfo, *, key: str | int | None = None) -> Generator[None, None, None]:
    """Context manager that sets the path of all GraphQL errors raised during its context."""
    try:
        yield
    except GraphQLError as error:
        if error.path is None:
            if key is not None:
                error.path = info.path.add_key(key).as_list()
            else:
                error.path = info.path.as_list()
        raise


def check_directives(directives: Iterable[Directive] | None, *, location: DirectiveLocation) -> None:
    """Check that given directives are allowed in the given location."""
    if directives is None:
        return

    for directive in directives:
        if location not in directive.__locations__:
            raise DirectiveLocationError(directive=directive, location=location)


def validate_get_request_operation(document: DocumentNode, operation_name: str | None = None) -> None:
    """Validates that the operation in the document can be executed in an HTTP GET request."""
    operation_definitions: list[OperationDefinitionNode] = [
        definition_node
        for definition_node in document.definitions
        if isinstance(definition_node, OperationDefinitionNode)
    ]

    if len(operation_definitions) == 0:
        raise GraphQLGetRequestNoOperationError

    if operation_name is None:
        if len(operation_definitions) != 1:
            raise GraphQLGetRequestMultipleOperationsNoOperationNameError

        if operation_definitions[0].operation != OperationType.QUERY:
            raise GraphQLGetRequestNonQueryOperationError

        return

    for operation in operation_definitions:
        if operation.name is None:
            continue

        if operation.name.value != operation_name:
            continue

        if operation.operation != OperationType.QUERY:
            raise GraphQLGetRequestNonQueryOperationError

        return

    raise GraphQLGetRequestOperationNotFoundError(operation_name=operation_name)

1	from __future__ import annotations	9✔
2
3	from contextlib import contextmanager	9✔
4	from typing import TYPE_CHECKING, Any, TypeGuard, TypeVar	9✔
5
6	from django.db.models import ForeignKey	9✔
7	from graphql import (	9✔
8	DocumentNode,
9	FieldNode,
10	GraphQLEnumType,
11	GraphQLError,
12	GraphQLIncludeDirective,
13	GraphQLInputObjectType,
14	GraphQLInterfaceType,
15	GraphQLObjectType,
16	GraphQLScalarType,
17	GraphQLSkipDirective,
18	GraphQLUnionType,
19	OperationDefinitionNode,
20	OperationType,
21	get_argument_values,
22	get_directive_values,
23	)
24
25	from undine.exceptions import (	9✔
26	DirectiveLocationError,
27	GraphQLGetRequestMultipleOperationsNoOperationNameError,
28	GraphQLGetRequestNonQueryOperationError,
29	GraphQLGetRequestNoOperationError,
30	GraphQLGetRequestOperationNotFoundError,
31	)
32	from undine.utils.text import to_snake_case	9✔
33
34	if TYPE_CHECKING:
35	from collections.abc import Generator, Iterable
36
37	from graphql import (
38	DirectiveLocation,
39	DocumentNode,
40	GraphQLList,
41	GraphQLNonNull,
42	GraphQLOutputType,
43	GraphQLWrappingType,
44	SelectionNode,
45	)
46	from graphql.execution.values import NodeWithDirective
47
48	from undine import Field, GQLInfo
49	from undine.directives import Directive
50	from undine.typing import ModelField
51
52
53	__all__ = [	9✔
54	"check_directives",
55	"get_arguments",
56	"get_queried_field_name",
57	"get_underlying_type",
58	"is_connection",
59	"is_edge",
60	"is_node_interface",
61	"is_page_info",
62	"is_relation_id",
63	"is_subscription_operation",
64	"should_skip_node",
65	"with_graphql_error_path",
66	]
67
68
69	TGraphQLType = TypeVar(	9✔
70	"TGraphQLType",
71	GraphQLScalarType,
72	GraphQLObjectType,
73	GraphQLInterfaceType,
74	GraphQLUnionType,
75	GraphQLEnumType,
76	GraphQLInputObjectType,
77	)
78
79
80	# Getters
81
82
83	def get_underlying_type(	9✔
84	gql_type: (
85	TGraphQLType
86	\| GraphQLList[TGraphQLType]
87	\| GraphQLList[GraphQLNonNull[TGraphQLType]]
88	\| GraphQLNonNull[TGraphQLType]
89	\| GraphQLNonNull[GraphQLList[TGraphQLType]]
90	\| GraphQLNonNull[GraphQLList[GraphQLNonNull[TGraphQLType]]]
91	\| GraphQLWrappingType[TGraphQLType]
92	),
93	) -> TGraphQLType:
94	while hasattr(gql_type, "of_type"):	9✔
95	gql_type = gql_type.of_type	9✔
96	return gql_type	9✔
97
98
99	def get_arguments(info: GQLInfo) -> dict[str, Any]:	9✔
100	"""Get input arguments for the current field from the GraphQL resolve info."""
101	graphql_field = info.parent_type.fields[info.field_name]	9✔
102	return get_argument_values(graphql_field, info.field_nodes[0], info.variable_values)	9✔
103
104
105	def get_queried_field_name(original_name: str, info: GQLInfo) -> str:	9✔
106	"""Get the name of a field in the current query."""
107	return original_name if info.path.key == info.field_name else info.path.key # type: ignore[return-value]	9✔
108
109
110	async def pre_evaluate_request_user(info: GQLInfo) -> None:	9✔
111	"""
112	Fetches the request user from the context and caches it to the request.
113	This is a workaround when current user is required in an async event loop,
114	but the function itself is not async.
115	"""
116	# '_current_user' would be set by 'django.contrib.auth.middleware.get_user' when calling 'request.user'
117	info.context._cached_user = await info.context.auser() # type: ignore[attr-defined] # noqa: SLF001	9✔
118
119
120	# Predicates
121
122
123	def is_connection(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:	9✔
124	return (	9✔
125	isinstance(field_type, GraphQLObjectType)
126	and field_type.name.endswith("Connection")
127	and "pageInfo" in field_type.fields
128	and "edges" in field_type.fields
129	)
130
131
132	def is_edge(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:	9✔
133	return (	9✔
134	isinstance(field_type, GraphQLObjectType)
135	and field_type.name.endswith("Edge")
136	and "cursor" in field_type.fields
137	and "node" in field_type.fields
138	)
139
140
141	def is_node_interface(field_type: GraphQLOutputType) -> TypeGuard[GraphQLInterfaceType]:	9✔
142	return (	9✔
143	isinstance(field_type, GraphQLInterfaceType) # comment here for better formatting
144	and field_type.name == "Node"
145	and "id" in field_type.fields
146	)
147
148
149	def is_page_info(field_type: GraphQLOutputType) -> TypeGuard[GraphQLObjectType]:	9✔
150	return (	9✔
151	isinstance(field_type, GraphQLObjectType)
152	and field_type.name == "PageInfo"
153	and "hasNextPage" in field_type.fields
154	and "hasPreviousPage" in field_type.fields
155	and "startCursor" in field_type.fields
156	and "endCursor" in field_type.fields
157	)
158
159
160	def is_typename_metafield(field_node: SelectionNode) -> TypeGuard[FieldNode]:	9✔
161	if not isinstance(field_node, FieldNode):	9✔
162	return False	9✔
163	return field_node.name.value.lower() == "__typename"	9✔
164
165
166	def is_relation_id(field: ModelField, field_node: FieldNode) -> TypeGuard[Field]:	9✔
167	return isinstance(field, ForeignKey) and field.get_attname() == to_snake_case(field_node.name.value)	9✔
168
169
170	def is_subscription_operation(document: DocumentNode) -> bool:	9✔
171	if len(document.definitions) != 1:	9✔
172	return False	9✔
173
174	operation_definition = document.definitions[0]	9✔
175	if not isinstance(operation_definition, OperationDefinitionNode):	9✔
NEW 176	return False	×
177
178	return operation_definition.operation == OperationType.SUBSCRIPTION	9✔
179
180
181	def should_skip_node(node: NodeWithDirective, variable_values: dict[str, Any]) -> bool:	9✔
182	skip_args = get_directive_values(GraphQLSkipDirective, node, variable_values)	9✔
183	if skip_args is not None and skip_args["if"] is True:	9✔
184	return True	9✔
185
186	include_args = get_directive_values(GraphQLIncludeDirective, node, variable_values)	9✔
187	return include_args is not None and include_args["if"] is False	9✔
188
189
190	# Misc.
191
192
193	@contextmanager	9✔
194	def with_graphql_error_path(info: GQLInfo, *, key: str \| int \| None = None) -> Generator[None, None, None]:	9✔
195	"""Context manager that sets the path of all GraphQL errors raised during its context."""
196	try:	9✔
197	yield	9✔
198	except GraphQLError as error:	9✔
199	if error.path is None:	9✔
200	if key is not None:	9✔
201	error.path = info.path.add_key(key).as_list()	9✔
202	else:
203	error.path = info.path.as_list()	9✔
204	raise	9✔
205
206
207	def check_directives(directives: Iterable[Directive] \| None, *, location: DirectiveLocation) -> None:	9✔
208	"""Check that given directives are allowed in the given location."""
209	if directives is None:	9✔
210	return	×
211
212	for directive in directives:	9✔
213	if location not in directive.__locations__:	9✔
214	raise DirectiveLocationError(directive=directive, location=location)	9✔
215
216
217	def validate_get_request_operation(document: DocumentNode, operation_name: str \| None = None) -> None:	9✔
218	"""Validates that the operation in the document can be executed in an HTTP GET request."""
219	operation_definitions: list[OperationDefinitionNode] = [	9✔
220	definition_node
221	for definition_node in document.definitions
222	if isinstance(definition_node, OperationDefinitionNode)
223	]
224
225	if len(operation_definitions) == 0:	9✔
226	raise GraphQLGetRequestNoOperationError	×
227
228	if operation_name is None:	9✔
229	if len(operation_definitions) != 1:	9✔
230	raise GraphQLGetRequestMultipleOperationsNoOperationNameError	×
231
232	if operation_definitions[0].operation != OperationType.QUERY:	9✔
233	raise GraphQLGetRequestNonQueryOperationError	9✔
234
235	return	9✔
236
237	for operation in operation_definitions:	×
238	if operation.name is None:	×
239	continue	×
240
241	if operation.name.value != operation_name:	×
242	continue	×
243
244	if operation.operation != OperationType.QUERY:	×
245	raise GraphQLGetRequestNonQueryOperationError	×
246
247	return	×
248
249	raise GraphQLGetRequestOperationNotFoundError(operation_name=operation_name)	×

MrThearMan / undine / 16081423623

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous