fixed custom directives

This commit is contained in:
MardanovTimur 2019-03-22 22:51:19 +03:00
parent 85d45414f9
commit 87a98fc6f8
8 changed files with 127 additions and 0 deletions

View File

View File

0
examples/neomodel/nodes Normal file
View File

View File

28
examples/neomodel/test.py Normal file
View File

@ -0,0 +1,28 @@
def find_brakes(self, recursion):
pass
REGEXES = [
r'((?P<recursive_points>\*+)(?P<model_name>\s?\w+\s*)(?P<body>(\{|.*|\n*|\})*))',
r'(\*node\s)*(\{(((?>[^{}]+)|(?R))*)\})',
]
query = """
query recursion_point {
module {
id
name
*module {
edges {
node {
id
}
}
}
}
}
"""
if __name__ == "__main__":

View File

@ -0,0 +1,24 @@
from .middleware import BaseCustomDirective
from graphql.type.definition import GraphQLArgument, GraphQLNonNull
from graphql.type.scalars import GraphQLString
class DefaultDirective(BaseCustomDirective):
"""
Default to given value if None
"""
@staticmethod
def get_args():
return {
'to': GraphQLArgument(
type=GraphQLNonNull(GraphQLString),
description='Value to default to',
),
}
@staticmethod
def process(value, directive, root, info, **kwargs):
if value is None:
to_argument = [arg for arg in directive.arguments if arg.name.value == 'to'][0]
return to_argument.value.value
return value

View File

@ -0,0 +1,68 @@
import six
from graphql import DirectiveLocation, GraphQLDirective
from promise import Promise
class DirectivesMiddleware(object):
def resolve(self, next, root, info, **kwargs):
result = next(root, info, **kwargs)
return result.then(
lambda resolved: self.__process_value(resolved, root, info, **kwargs),
lambda error: Promise.rejected(error)
)
def __process_value(self, value, root, info, **kwargs):
field = info.field_asts[0]
# if field has not directive - return model-value
if not field.directives:
return value
new_value = value
# for each by field.directives
for directive in field.directives:
directive_class = CustomDirectiveMeta.REGISTRY.get(directive.name.value)
if directive_class:
# if directive class found
new_value = directive_class.process(new_value, directive, root, info, **kwargs)
return new_value
class CustomDirectiveMeta(type):
REGISTRY = {}
def __new__(mcs, name, bases, attrs):
newclass = super(CustomDirectiveMeta, mcs).__new__(mcs, name, bases, attrs)
if name != 'BaseCustomDirective':
mcs.register(newclass)
return newclass
@classmethod
def register(mcs, target):
mcs.REGISTRY[target.get_name()] = target
@classmethod
def get_all_directives(cls):
return [d() for d in cls.REGISTRY.values()]
class BaseCustomDirective(six.with_metaclass(CustomDirectiveMeta, GraphQLDirective)):
__metaclass__ = CustomDirectiveMeta
def __init__(self):
super(BaseCustomDirective, self).__init__(
name=self.get_name(),
description=self.__doc__,
args=self.get_args(),
locations=[
DirectiveLocation.FIELD
]
)
@classmethod
def get_name(cls):
return cls.__name__.replace('Directive', '').lower()
@staticmethod
def get_args():
return {}
from .directives import * # define the directives

View File

@ -17,6 +17,7 @@ from graphql.execution import ExecutionResult
from graphql.type.schema import GraphQLSchema from graphql.type.schema import GraphQLSchema
from .settings import graphene_settings from .settings import graphene_settings
from .middleware import DirectivesMiddleware
class HttpError(Exception): class HttpError(Exception):
@ -84,6 +85,7 @@ class GraphQLView(View):
middleware = graphene_settings.MIDDLEWARE middleware = graphene_settings.MIDDLEWARE
self.schema = self.schema or schema self.schema = self.schema or schema
middleware = self.get_directive_middleware()
if middleware is not None: if middleware is not None:
self.middleware = list(instantiate_middleware(middleware)) self.middleware = list(instantiate_middleware(middleware))
self.executor = executor self.executor = executor
@ -98,6 +100,11 @@ class GraphQLView(View):
), "A Schema is required to be provided to GraphQLView." ), "A Schema is required to be provided to GraphQLView."
assert not all((graphiql, batch)), "Use either graphiql or batch processing" assert not all((graphiql, batch)), "Use either graphiql or batch processing"
def get_directive_middleware(self, custom=True):
return [DirectivesMiddleware, ]
# noinspection PyUnusedLocal # noinspection PyUnusedLocal
def get_root_value(self, request): def get_root_value(self, request):
return self.root_value return self.root_value