[컴][파이썬] Django 에서 graphQL library 의 동작(graphene_django)






Django 에서 graphQL library 의 동작(graphene_django)


아래와 같은 graphQL 이 client --> server 로 날라가면
query{
  user{
    id
  }
}

서버에서 개발자가 미리 정해놓은 schema 에 맞춰서 response 를 보내주게 된다. 보통 django 에서는 이것을 schema.py 에 만들어 놓는다.

위의 query 에 대한 응답은 아래와 같다.
{
  "data": {
    "users": [
      {
        "id": "1"
      }
    ]
  }
}


대략적은 동작은 아래글을 통해 파악하자.




Source flow

위와 같은 response 가 오기 위해서 django 의 graphQL library 는 어떤 과정을 거치는지 source 를 살펴보자.


# my_django_project/api/schema.py
import graphene
from graphene_django import DjangoObjectType
from api.models import UserModel

class User(DjangoObjectType):
    class Meta:
        model = UserModel

class Query(graphene.ObjectType):
    users = graphene.List(User)

    @graphene.resolve_only_args
    def resolve_users(self):
        return UserModel.objects.all()

schema = graphene.Schema(query=Query)




# my_django_project/api/urls.py
...
from graphene_django.views import GraphQLView

urlpatterns = [
    
    url(r'^graphql', GraphQLView.as_view(graphiql=True)),
    
]


---- libraries


# django/views/generic/base.py
class View(object):
    ...

    @classonlymethod
    def as_view(cls, **initkwargs):
        ...

        def view(request, *args, **kwargs):
            self = cls(**initkwargs)
            if hasattr(self, 'get') and not hasattr(self, 'head'):
                self.head = self.get
            self.request = request
            self.args = args
            self.kwargs = kwargs
            return self.dispatch(request, *args, **kwargs)



# graphene_django/views.py

class GraphQLView(View):
    graphiql_version = '0.7.8'
    graphiql_template = 'graphene/graphiql.html'

    schema = None
    graphiql = False
    executor = None
    middleware = None
    root_value = None
    pretty = False

    def __init__(self, schema=None, executor=None, middleware=None, root_value=None, graphiql=False, pretty=False):
        if not schema:
            schema = graphene_settings.SCHEMA

        if middleware is None:
            middleware = graphene_settings.MIDDLEWARE

        self.schema = schema
        if middleware is not None:
            self.middleware = list(instantiate_middleware(middleware))
        self.executor = executor
        self.root_value = root_value
        self.pretty = pretty
        self.graphiql = graphiql

        assert isinstance(self.schema, GraphQLSchema), 'A Schema is required to be provided to GraphQLView.'

    ...
    @method_decorator(ensure_csrf_cookie)
    def dispatch(self, request, *args, **kwargs):
        try:
            if request.method.lower() not in ('get', 'post'):
                raise HttpError(HttpResponseNotAllowed(['GET', 'POST'], 'GraphQL only supports GET and POST requests.'))

            data = self.parse_body(request)
            show_graphiql = self.graphiql and self.can_display_graphiql(request, data)

            query, variables, operation_name = self.get_graphql_params(request, data)

            execution_result = self.execute_graphql_request(
                request,
                data,
                query, # 'query{ user {id }}'
                variables,
                operation_name,
                show_graphiql
            )

            if execution_result:
                ...

                result = self.json_encode(request, response, pretty=show_graphiql)
            else:
                result = None

            if show_graphiql:
                return self.render_graphiql(
                    request,
                    graphiql_version=self.graphiql_version,
                    query=query or '',
                    variables=variables or '',
                    operation_name=operation_name or '',
                    result=result or ''
                )

            return HttpResponse(
                status=status_code,
                content=result,
                content_type='application/json'
            )

        except HttpError as e:
            ...
            return response

    def execute_graphql_request(self, request, data, query, variables, operation_name, show_graphiql=False):
        ...
        source = Source(query, name='GraphQL request')

        try:
            document_ast = parse(source)
            validation_errors = validate(self.schema, document_ast)
            if validation_errors:
                return ExecutionResult(
                    errors=validation_errors,
                    invalid=True,
                )
        except Exception as e:
            return ExecutionResult(errors=[e], invalid=True)
        
        ...

        try:
            return self.execute(
                document_ast,
                root_value=self.get_root_value(request),
                variable_values=variables,
                operation_name=operation_name,
                context_value=self.get_context(request),
                middleware=self.get_middleware(request),
                executor=self.executor,
            )
        except Exception as e:
            return ExecutionResult(errors=[e], invalid=True)

    def execute(self, *args, **kwargs):
        return execute(self.schema, *args, **kwargs)


# graphql/execution/executor.py
def execute(schema, document_ast, root_value=None, context_value=None,
            variable_values=None, operation_name=None, executor=None,
            return_promise=False, middleware=None):
    assert schema, 'Must provide schema'
    assert isinstance(schema, GraphQLSchema), (
        'Schema must be an instance of GraphQLSchema. Also ensure that there are ' +
        'not multiple versions of GraphQL installed in your node_modules directory.'
    )
    if middleware:
        if not isinstance(middleware, MiddlewareManager):
            middleware = MiddlewareManager(*middleware)

        ...

    if executor is None:
        executor = SyncExecutor()

    context = ExecutionContext(
        schema,
        document_ast,
        root_value,
        context_value,
        variable_values,
        operation_name,
        executor,
        middleware
    )

    def executor(resolve, reject):
        return resolve(execute_operation(context, context.operation, root_value))

    def on_rejected(error):
        context.errors.append(error)
        return None

    def on_resolve(data):
        return ExecutionResult(data=data, errors=context.errors)

    promise = Promise(executor).catch(on_rejected).then(on_resolve)
    ...
    

def execute_operation(exe_context, operation, root_value):
    type = get_operation_root_type(exe_context.schema, operation)
    fields = collect_fields(
        exe_context,
        type,
        operation.selection_set,
        DefaultOrderedDict(list),
        set()
    )

    if operation.operation == 'mutation':
        return execute_fields_serially(exe_context, type, root_value, fields)

    return execute_fields(exe_context, type, root_value, fields)


def execute_fields(exe_context, parent_type, source_value, fields):
    contains_promise = False

    final_results = OrderedDict()

    for response_name, field_asts in fields.items():
        result = resolve_field(exe_context, parent_type, source_value, field_asts)
        if result is Undefined:
            continue

        final_results[response_name] = result
        if is_promise(result):
            contains_promise = True

def resolve_field(exe_context, parent_type, source, field_asts):
    field_ast = field_asts[0]
    # at this example, field_name is 'users'
    field_name = field_ast.name.value

    field_def = get_field_def(exe_context.schema, parent_type, field_name)
    if not field_def:
        return Undefined

    # field_def is GraphQLField type
    return_type = field_def.type  
    # field_def.resolver is api.schema.Query.resolve_users
    resolve_fn = field_def.resolver or default_resolve_fn

    # We wrap the resolve_fn from the middleware
    resolve_fn_middleware = exe_context.get_field_resolver(resolve_fn)

    # Build a dict of arguments from the field.arguments AST, using the variables scope to
    # fulfill any variable references.
    args = exe_context.get_argument_values(field_def, field_ast)

    # The resolve function's optional third argument is a context value that
    # is provided to every resolve function within an execution. It is commonly
    # used to represent an authenticated user, or request-specific caches.
    context = exe_context.context_value

    # The resolve function's optional third argument is a collection of
    # information about the current execution state.
    info = ResolveInfo(
        field_name,
        field_asts,
        return_type,
        parent_type,
        schema=exe_context.schema,
        fragments=exe_context.fragments,
        root_value=exe_context.root_value,
        operation=exe_context.operation,
        variable_values=exe_context.variable_values,
    )

    executor = exe_context.executor
    result = resolve_or_error(resolve_fn_middleware, source, args, context, info, executor)
    ...






댓글 없음:

댓글 쓰기