graphql-python/graphene
1
1
Fork 0
mirror of https://github.com/graphql-python/graphene.git synced 2024-11-23 01:56:54 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update schema.py

Browse Source
This commit is contained in:
Aryan Iyappan 2021-08-20 15:58:43 +05:30 committed by GitHub
parent 7d890bf915
commit 946c2a3807
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 0 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

236
graphene/types/schema.py
View File

@ -391,239 +391,3 @@ class TypeMap(dict):
return graphql_type return graphql_type
return type_ return type_
class UnforgivingExecutionContext(ExecutionContext):
"""An execution context which doesn't swallow exceptions.
The only difference between this execution context and the one it inherits from is
that ``except Exception`` is commented out within ``resolve_field_value_or_error``.
By removing that exception handling, only ``GraphQLError``'s are caught.
"""
def resolve_field_value_or_error(
self, field_def, field_nodes, resolve_fn, source, info
):
"""Resolve field to a value or an error.
Isolates the "ReturnOrAbrupt" behavior to not de-opt the resolve_field()
method. Returns the result of resolveFn or the abrupt-return Error object.
For internal use only.
"""
try:
# Build a dictionary of arguments from the field.arguments AST, using the
# variables scope to fulfill any variable references.
args = get_argument_values(field_def, field_nodes[0], self.variable_values)
# Note that contrary to the JavaScript implementation, we pass the context
# value as part of the resolve info.
result = resolve_fn(source, info, **args)
if self.is_awaitable(result):
# noinspection PyShadowingNames
async def await_result():
try:
return await result
except GraphQLError as error:
return error
# except Exception as error:
# return GraphQLError(str(error), original_error=error)
# Yes, this is commented out code. It's been intentionally
# _not_ removed to show what has changed from the original
# implementation.
return await_result()
return result
except GraphQLError as error:
return error
# except Exception as error:
# return GraphQLError(str(error), original_error=error)
# Yes, this is commented out code. It's been intentionally _not_
# removed to show what has changed from the original implementation.
def complete_value_catching_error(
self, return_type, field_nodes, info, path, result
):
"""Complete a value while catching an error.
This is a small wrapper around completeValue which detects and logs errors in
the execution context.
"""
try:
if self.is_awaitable(result):
async def await_result():
value = self.complete_value(
return_type, field_nodes, info, path, await result
)
if self.is_awaitable(value):
return await value
return value
completed = await_result()
else:
completed = self.complete_value(
return_type, field_nodes, info, path, result
)
if self.is_awaitable(completed):
# noinspection PyShadowingNames
async def await_completed():
try:
return await completed
# CHANGE WAS MADE HERE
# ``GraphQLError`` was swapped in for ``except Exception``
except GraphQLError as error:
self.handle_field_error(error, field_nodes, path, return_type)
return await_completed()
return completed
# CHANGE WAS MADE HERE
# ``GraphQLError`` was swapped in for ``except Exception``
except GraphQLError as error:
self.handle_field_error(error, field_nodes, path, return_type)
return None
class Schema:
"""Schema Definition.
A Graphene Schema can execute operations (query, mutation, subscription) against the defined
types. For advanced purposes, the schema can be used to lookup type definitions and answer
questions about the types through introspection.
Args:
query (Type[ObjectType]): Root query *ObjectType*. Describes entry point for fields to *read*
data in your Schema.
mutation (Optional[Type[ObjectType]]): Root mutation *ObjectType*. Describes entry point for
fields to *create, update or delete* data in your API.
subscription (Optional[Type[ObjectType]]): Root subscription *ObjectType*. Describes entry point
for fields to receive continuous updates.
types (Optional[List[Type[ObjectType]]]): List of any types to include in schema that
may not be introspected through root types.
directives (List[GraphQLDirective], optional): List of custom directives to include in the
GraphQL schema. Defaults to only include directives defined by GraphQL spec (@include
and @skip) [GraphQLIncludeDirective, GraphQLSkipDirective].
auto_camelcase (bool): Fieldnames will be transformed in Schema's TypeMap from snake_case
to camelCase (preferred by GraphQL standard). Default True.
"""
def __init__(
self,
query=None,
mutation=None,
subscription=None,
types=None,
directives=None,
auto_camelcase=True,
):
self.query = query
self.mutation = mutation
self.subscription = subscription
type_map = TypeMap(
query, mutation, subscription, types, auto_camelcase=auto_camelcase
)
self.graphql_schema = GraphQLSchema(
type_map.query,
type_map.mutation,
type_map.subscription,
type_map.types,
directives,
)
def __str__(self):
return print_schema(self.graphql_schema)
def __getattr__(self, type_name):
"""
This function let the developer select a type in a given schema
by accessing its attrs.
Example: using schema.Query for accessing the "Query" type in the Schema
"""
_type = self.graphql_schema.get_type(type_name)
if _type is None:
raise AttributeError(f'Type "{type_name}" not found in the Schema')
if isinstance(_type, GrapheneGraphQLType):
return _type.graphene_type
return _type
def lazy(self, _type):
return lambda: self.get_type(_type)
def execute(self, *args, **kwargs):
"""Execute a GraphQL query on the schema.
Use the `graphql_sync` function from `graphql-core` to provide the result
for a query string. Most of the time this method will be called by one of the Graphene
:ref:`Integrations` via a web request.
Args:
request_string (str or Document): GraphQL request (query, mutation or subscription)
as string or parsed AST form from `graphql-core`.
root_value (Any, optional): Value to use as the parent value object when resolving
root types.
context_value (Any, optional): Value to be made available to all resolvers via
`info.context`. Can be used to share authorization, dataloaders or other
information needed to resolve an operation.
variable_values (dict, optional): If variables are used in the request string, they can
be provided in dictionary form mapping the variable name to the variable value.
operation_name (str, optional): If multiple operations are provided in the
request_string, an operation name must be provided for the result to be provided.
middleware (List[SupportsGraphQLMiddleware]): Supply request level middleware as
defined in `graphql-core`.
execution_context_class (ExecutionContext, optional): The execution context class
to use when resolving queries and mutations.
Returns:
:obj:`ExecutionResult` containing any data and errors for the operation.
"""
kwargs = normalize_execute_kwargs(kwargs)
return graphql_sync(self.graphql_schema, *args, **kwargs)
async def execute_async(self, *args, **kwargs):
"""Execute a GraphQL query on the schema asynchronously.
Same as `execute`, but uses `graphql` instead of `graphql_sync`.
"""
kwargs = normalize_execute_kwargs(kwargs)
return await graphql(self.graphql_schema, *args, **kwargs)
async def subscribe(self, query, *args, **kwargs):
"""Execute a GraphQL subscription on the schema asynchronously."""
# Do parsing
try:
document = parse(query)
except GraphQLError as error:
return ExecutionResult(data=None, errors=[error])
# Do validation
validation_errors = validate(self.graphql_schema, document)
if validation_errors:
return ExecutionResult(data=None, errors=validation_errors)
# Execute the query
kwargs = normalize_execute_kwargs(kwargs)
return await subscribe(self.graphql_schema, document, *args, **kwargs)
def introspect(self):
introspection = self.execute(introspection_query)
if introspection.errors:
raise introspection.errors[0]
return introspection.data
def normalize_execute_kwargs(kwargs):
"""Replace alias names in keyword arguments for graphql()"""
if "root" in kwargs and "root_value" not in kwargs:
kwargs["root_value"] = kwargs.pop("root")
if "context" in kwargs and "context_value" not in kwargs:
kwargs["context_value"] = kwargs.pop("context")
if "variables" in kwargs and "variable_values" not in kwargs:
kwargs["variable_values"] = kwargs.pop("variables")
if "operation" in kwargs and "operation_name" not in kwargs:
kwargs["operation_name"] = kwargs.pop("operation")
return kwargs