关于python:graphenedjango-源码走读

前言

最近我的项目在应用 GraphQL 然而对其外部实现不是很理解, 能自在的指定查问字段以及自定义返回格局, 这样的灵便度能够说十分高了, 自由度越高也就意味着前面的简单, 查问百度真的找不到一篇能看的, 遂产生了一些想法:

• GraphQL 到底是个啥
• GraphQL 何如与 django 集成的
• 又是如何驱动 django orm 来执行查问操作的

刚开始间接用 IDE 看源码了, 看来一会发现不对劲, 外部蕴含了各种接口和协定又对应的实现又扩散在不同的库中, 不是独立的小我的项目, 想明确 GraphQL 和 db 是如何买通的, 须要对 orm 相熟, 而后再打断点反复看几次能力理出点思路, 间接看源码波及的文件和流程太多, 须要一一击破, 层层递进. 目前的思路是:

• 理解 graphsql 的根本应用和标准, 各个语言的库都是对此标准的实现, 明确各个名词的意思.
• 大抵理解这是三个外围库的应用 graphene_django,graphene,graphql(有包含 graphql-core,graphql-relay). 本次走读中三者是套娃构造. 都是前者对后者的根底上封装和调用.
• 再看 graphene_django 是如何缝合 graphene 和 django, 在 django 中 graphene_django 库和 graphene 是混合应用的, 有些中央档次不清
• 再串起来看 orm 在整个流程中是如何被传递和执行的, 其中次要关注 graphql 中是如何根据传入的参数去执行函数
• 浏览源码过程中依赖的相干常识: graphql 的基础知识;promis 的链式调用;

环境配置

graphene==2.1.8
graphene-django==2.11.0
graphql-core==2.3.2
graphql-relay==2.0.1

Python 行断点

Python 行断点
mapp/cluster/types.py
python3.8/site-packages/graphene_django/views.py
views.py:68
views.py:280
views.py:245
views.py:165
views.py:122
python3.8/site-packages/graphql/backend/core.py
core.py:32
core.py:45
python3.8/site-packages/graphql/execution/executor.py
executor.py:144
executor.py:531
executor.py:452
executor.py:365
executor.py:113
executor.py:59
python3.8/site-packages/graphql/execution/utils.py
utils.py:59

一些记录

lib/python3.8/site-packages/graphql/execution/executor.py:144
graphql.execution.executor.execute
Promise.resolve(None).then(promise_executor).catch(on_rejected).then(on_resolve)
promise 的链式调用

/Users/liuguojin/gitlab/venv_2/lib/python3.8/site-packages/graphene_django/debug/middleware.py:51
graphene_django.debug.middleware.DjangoDebugMiddleware
gql 获取 db 的 context, 并将查问工作形成的 promise 退出到查问的工作列表中.

graphql_relay.node.node.from_global_id
graphql_relay.utils.unbase64
这里做的 base64 转换.

graphene_django.types.DjangoObjectType.get_node
到这里应用 node 模式的默认形式取得查问语句.
graphene.relay.node.Node.node_resolver
次要和 db 对接的中央
graphql.backend.core.GraphQLCoreBackend

setting.DEBUG 最好设置为 Flase, 不然会把上面的类引入, 没有必要看
graphene_django.debug.middleware.DjangoDebugContext
graphene_django.debug.middleware.DjangoDebugMiddleware

GraphQLView 外围代码

graphene_django.views.GraphQLView , 继承了 django.views.generic.base.View 实现了对 view 的处理函数 dispatch, get_response.
其中 execute_graphql_request 实现了调用 graphql 的逻辑, 上面是对 GraphQLView 的一些注解, 这部分相熟 django 的都了解.

class GraphQLView(View):
    graphiql_version = "0.14.0"
    graphiql_template = "graphene/graphiql.html"
    react_version = "16.8.6"

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

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

        if backend is None:
            backend = get_default_backend()

        if middleware is None:
            middleware = graphene_settings.MIDDLEWARE

        self.schema = self.schema or schema
        if middleware is not None:
            if isinstance(middleware, MiddlewareManager):
                self.middleware = middleware
            else:
                self.middleware = list(instantiate_middleware(middleware))
        self.executor = executor
        self.root_value = root_value
        self.pretty = self.pretty or pretty
        self.graphiql = self.graphiql or graphiql
        self.batch = self.batch or batch
        self.backend = backend

    @method_decorator(ensure_csrf_cookie)
    def dispatch(self, request, *args, **kwargs):  # 重写父类的 dispatch 函数, 调用本人实现的 get_response
        try:
            # 删减了校验和其余不重要的分支代码
            data = self.parse_body(request)
            show_graphiql = self.graphiql and self.can_display_graphiql(request, data)
            result, status_code = self.get_response(request, data, show_graphiql)
            return HttpResponse(status=status_code, content=result, content_type="application/json")
        except HttpError as e:
            response = e.response
            response["Content-Type"] = "application/json"
            response.content = self.json_encode(request, {"errors": [self.format_error(e)]}
            )
            return response

    def get_response(self, request, data, show_graphiql=False):  # 解决 gql
        query, variables, operation_name, id = self.get_graphql_params(request, data)  # 获取参数
        
        # 要害语句
        execution_result = self.execute_graphql_request(request, data, query, variables, operation_name, show_graphiql)
        status_code = 200
        if execution_result:
            # 删减了局部代码
            result = self.json_encode(request, response, pretty=show_graphiql)
        else:
            result = None
        return result, status_code

    def execute_graphql_request(self, request, data, query, variables, operation_name, show_graphiql=False):
        # 对接 graphql, 满足其 execute 的调用条件, 你能够在这里看到十分类似的代码:
        # python3.8/site-packages/graphql/graphql.py(graphql.graphql.execute_graphql)
        try:
            backend = self.get_backend(request)
            # 获取 graphql.backend.base.GraphQLDocument
            # schema(注册表)是以后利用反对的操作合集
            document = backend.document_from_string(self.schema, query) # 初始化 GraphQLDocument 来适配 graphql 库
        except Exception as e:
            return ExecutionResult(errors=[e], invalid=True)
        try:
            extra_options = {}
            if self.executor:
                # We only include it optionally since
                # executor is not a valid argument in all backends
                extra_options["executor"] = self.executor

            # 开始执行查问, 之后就得浏览 graphql 的源码
            return document.execute(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),
                **extra_options
            )
        except Exception as e:
            return ExecutionResult(errors=[e], invalid=True)

    def json_encode(self, request, d, pretty=False):  # 格式化返回
        if not (self.pretty or pretty) and not request.GET.get("pretty"):
            return json.dumps(d, separators=(",", ":"))

        return json.dumps(d, sort_keys=True, indent=2, separators=(",", ":"))

    def get_backend(self, request):
        return self.backend
    
    def parse_body(self, request):  # 获取申请参数
        content_type = self.get_content_type(request)
        if content_type == "application/graphql":
            return {"query": request.body.decode()}

    @staticmethod
    def get_graphql_params(request, data):
        # 获取参数
        query = request.GET.get("query") or data.get("query")
        variables = request.GET.get("variables") or data.get("variables")
        id = request.GET.get("id") or data.get("id")

        if variables and isinstance(variables, six.text_type):
            try:
                variables = json.loads(variables)
            except Exception:
                raise HttpError(HttpResponseBadRequest("Variables are invalid JSON."))

        operation_name = request.GET.get("operationName") or data.get("operationName")
        if operation_name == "null":
            operation_name = None
        return query, variables, operation_name, id

execution.executor.execute 连环套

graphql.execution.executor.execute 真正开始筹备执行查问的中央, 波及到了 promise 的用法
相熟 JavaScript 的会感觉很相熟, 就是 promise 的链式调用, 解决 callback 的形式. 代码中应用到的 promise 就是 promise 的 python 实现库.

def execute(
    schema,  # type: GraphQLSchema
    document_ast,  # type: Document
    root_value=None,  # type: Any
    context_value=None,  # type: Optional[Any]
    variable_values=None,  # type: Optional[Any]
    operation_name=None,  # type: Optional[str]
    executor=None,  # type: Any
    return_promise=False,  # type: bool
    middleware=None,  # type: Optional[Any]
    allow_subscriptions=False,  # type: bool
    **options  # type: Any
):
    # type: (...) -> Union[ExecutionResult, Promise[ExecutionResult]]

    if executor is None:
        executor = SyncExecutor()

    exe_context = ExecutionContext(
        schema,
        document_ast,
        root_value,
        context_value,
        variable_values or {},
        operation_name,
        executor,
        middleware,
        allow_subscriptions,
    )

    def promise_executor(v):
        # type: (Optional[Any]) -> Union[Dict, Promise[Dict], Observable]
        return execute_operation(exe_context, exe_context.operation, root_value)

    def on_rejected(error):
        # type: (Exception) -> None
        exe_context.errors.append(error)
        return None

    def on_resolve(data):
        # type: (Union[None, Dict, Observable]) -> Union[ExecutionResult, Observable]
        if isinstance(data, Observable):
            return data

        if not exe_context.errors:
            return ExecutionResult(data=data)

        return ExecutionResult(data=data, errors=exe_context.errors)

    # Promise 的链式调用, 大抵的意思就是实例化一个 Promise 而后执行 promise_executor 要是产生异样就执行 on_rejected, 没有异样就执行 on_resolve
    promise = (Promise.resolve(None).then(promise_executor).catch(on_rejected).then(on_resolve)
    )

    if not return_promise:
        exe_context.executor.wait_until_finished()  # 默认是不返回的 promise 对象的, 而是期待查问后果
        return promise.get()
    else:
        clean = getattr(exe_context.executor, "clean", None)
        if callable(clean):
            clean()

    return promise

promise 介绍

Promise 次要基于回调，Python asyncio 次要基于事件循环，两者相似，后者封装更多。
Promise 的实现过程，其次要应用了设计模式中的观察者模式：

• 通过 Promise.prototype.then 和 Promise.prototype.catch 办法将观察者办法注册到被观察者 Promise 对象中，同时返回一个新的 Promise 对象，以便能够链式调用。
• 被观察者治理外部 pending、fulfilled 和 rejected 的状态转变，同时通过构造函数中传递的 resolve 和 reject 办法以被动触发状态转变和告诉观察者。
  简略实现

field

而后剩下的注意力就得放在 graphql.execution.executor.resolve_field 这个函数上了, 这部分也很简单, 咱们先应用 graphene demo 开始. 其实上面的示例就能够脱离 django 和 graphene_django 的应用 graphene 给的 demo 来调试了.

from graphene import ObjectType, String, Schema

class Query(ObjectType):
    # this defines a Field `hello` in our Schema with a single Argument `name`
    hello = String(name=String(default_value="stranger"))
    goodbye = String()

    # our Resolver method takes the GraphQL context (root, info) as well as
    # Argument (name) for the Field and returns data for the query Response
    def resolve_hello(root, info, name):
        print('resolve_hello for debug')
        return f'Hello {name}!'

    def resolve_goodbye(root, info):
        print('resolve_goodbye for debug')
        return 'See ya!'

schema = Schema(query=Query)

# we can query for our field (with the default argument)
query_string = '{hello}'
result = schema.execute(query_string)
print(result.data['hello'])
# "Hello stranger!"

能够在这里 schema._type_map.Query.fields 看到咱们所有定义的处理函数, 下面的例子其解决如下, 依据 graphene 的约定处理函数前缀默认加resolve_, 这里是解决过后的.

{'hello': <graphql.type.definition.GraphQLField object at 0x10f180220>, 'goodbye': <graphql.type.definition.GraphQLField object at 0x10f180270>}

继续执行就会来到 graphql.execution.executor.execute 这里, 到这里执行前须要的所有必要参数都已筹备好, 接下就就得认真看看怎么执行的.

class GraphQLCoreBackend(GraphQLBackend):
    """GraphQLCoreBackend will return a document using the default
    graphql executor"""

    def __init__(self, executor=None):
        # type: (Optional[Any]) -> None
        self.execute_params = {"executor": executor}

    def document_from_string(self, schema, document_string):
        # type: (GraphQLSchema, Union[Document, str]) -> GraphQLDocument
        if isinstance(document_string, ast.Document):
            document_ast = document_string
            document_string = print_ast(document_ast) # 相似于路由匹配, 这里将决定执行哪个 resolve_XXXX
        else:
            assert isinstance(document_string, string_types), "The query must be a string"
            document_ast = parse(document_string)   # 相似于路由匹配, 这里将决定执行哪个 resolve_XXXX
        # document_ast 是一个 selection_set 组成的嵌套构造, 具体由 document_string 的复杂度决定, 每个档次都蕴含了须要执行的 Field, 这里要联合 get_field_def 来看.
        return GraphQLDocument(
            schema=schema,
            document_string=document_string,    
            document_ast=document_ast,  # 留神下这个参数
            execute=partial(    # 留神下这个参数
                execute_and_validate, schema, document_ast, **self.execute_params
            ),
            # 这里应用了 python 高阶函数中的 partial(偏函数), 将事后能够提供的参数先提供给之后将会调用的 execute 函数
        )

须要留神真正的处理函数是怎么在上面这些函数中传递(挺绕的):

schema-> document ->execute->execute_operation->collect_fields->execute_fields->resolve_field->field_def

到这里失去 result 而后关联上对应的 query 返回即可, 这里不再细说返回的代码.

    # graphql.execution.executor.resolve_field 蕴含了很多细节, 这里大略说两个.
    # 这一句很要害, 决定了到底是由哪个 field 来执行, parent_type 中蕴含了以后办法 Query 中所有的利用中定义的 query 办法.
    # graphql.type.schema.GraphQLSchema
    field_def = get_field_def(exe_context.schema, parent_type, field_name)
    # ........... 省略了局部代码
    executor = exe_context.executor # 抉择执行器
    # 将有所须要执行的函数(resolve_fn_middleware), 参数(args), 执行函数的执行器(executor)
    result = resolve_or_error(resolve_fn_middleware, source, info, args, executor)

执行器就是调用 fn 并把参数打包给 fn, 这也是为什么 resolve_hello 中的前两个参数是固定为 root, info 的起因.

class SyncExecutor(object):
    def wait_until_finished(self):
        # type: () -> None
        pass

    def clean(self):
        pass

    def execute(self, fn, *args, **kwargs):
        # type: (Callable, *Any, **Any) -> Any
        return fn(*args, **kwargs)

到这里 demo 的 query 操作根本就明确了, 接着去看看 query 是怎么调用 orm 的操作方法, 其实这部分操作次要得看 graphene-django 是如何封装的,
其逻辑的终点在这里 graphene_django.types.DjangoObjectType

class DjangoObjectType(ObjectType):
    @classmethod
    def __init_subclass_with_meta__(    # 相似元类的操作具体能够查看 __init_subclass__魔术办法, 作用是定制子类属性.
        cls,
        model=None,
        registry=None,
        skip_registry=False,
        only_fields=None,  # deprecated in favour of `fields`
        fields=None,
        exclude_fields=None,  # deprecated in favour of `exclude`
        exclude=None,
        filter_fields=None,
        filterset_class=None,
        connection=None,
        connection_class=None,
        use_connection=None,
        interfaces=(),
        convert_choices_to_enum=True,
        _meta=None,
        **options
    ):
        # 一堆限定条件和对外提供的参数来管制查问范畴, 形式等
        pass
    
    @classmethod
    def get_queryset(cls, queryset, info):
        return queryset

    @classmethod
    def get_node(cls, info, id):
        queryset = cls.get_queryset(cls._meta.model.objects, info)  # 这句将 django model 和 node 关联起来, 之后操作 node 即可.
        try:
            return queryset.get(pk=id)
        except cls._meta.model.DoesNotExist:
            return None

流程根本理清了, 然而 sql 执行的具体细节还须要看, 这就得回头看 graphene_django 中是如何包装 graphql-core 和 graphql-relay 以及 orm 的, 这部分真的太简短了, 不仅要对协定和各种名词相熟, 还要在三个库中来回跳, 后续有空再补充吧.

总结

看到这里的根本就该来个总结了，缓缓接触和应用 GraphQL 过程中我的心田是很顺当的。
GraphQL 的外围劣势就是一次性获取资源，看上去是炫酷，炫酷的骚操作有些简单的办法来实现也能够承受，但想配合这些简单的实现来插入些本人的非凡需要那就真的考验开发者的实力了, 走读源码的过程也验证了我之前的猜测,用起来有多不便, 实现起来就有多少简单 , 其中蕴含的接口和标准不是简略看看就能上手改变的, 小我的项目对优化没什么要求, 用到的操作仅限于实例的能够尝试，超过范畴的就进去了开发的深水区, 能看的文档都列举在 参考 中了。一句话:REST 真简略，GraphQL 如乱码, 将来还任重而道远。

参考

https://graphql.cn/learn/
https://spec.graphql.cn/
https://docs.graphene-python.org/en/latest/execution/

https://github.com/graphql-python
https://github.com/graphql-python/graphql-core
https://github.com/graphql-python/graphql-relay-py
https://mengera88.github.io/2017/05/18/Promise%E5%8E%9F%E7%90%86%E8%A7%A3%E6%9E%90/
https://github.com/syrusakbary/promise
https://www.jianshu.com/p/ca1dfc5b4b4f
https://www.zhihu.com/question/38596306 GraphQL 为何没有火起来?