Flask-GraphQL-SQLAlchemy 开发文档 | Develop Docs

作者 | Author：杨晗光

概述 | Overview

GraphQL是一个由facebook开发的API查询语言。对于前端开发者来说，它非常棒。但是有关于它的python开发文档却很少。所以我写了这个文档来帮助后来者快速搭建一个GraphQL服务。

GraphQL is a API query language developed by facebook. It’s pretty cool for frontend developers. But there are few documents about its python version. So I write this doc to help the later start quickly build a GraphQL service.

注意，这不是一个Python的新手教程，所以我将不会一步一步的教你怎么写代码。如果你发现某些部分难以理解，我想谷歌和百度是万能的。

Please note, this is not a beginner’s tutorial for Python, so I will not teach you step by step. If you find some parts are hard to understand, I think you can search the results by Baidu or Google.

这篇文档的原始版本是用于公司内部的纯英文说明，在这里抽取主要部分并翻译成中文后发布出来，如果有哪里感觉别扭还请见谅。

The original version of this document is for the company’s in-house instructions in pure English, I extracted the main parts and translated into Chinese and released, if you feel uncomfortable, please forgive me.

入门 | Getting Started

虚拟环境 | Virtual Environment

GraphQL同时支持python2.7和pythong3.4+，但是即将到来的更新至支持python3.6+。你最好用这个版本来创建虚拟环境。

GraphQL support both python2.7 and python3.4+, but the upcoming update only support python3.6+. You’d better build virtualenv from this version.

virtualenv -p python3.6 graphql
source graphql/bin/activate

安装依赖 | Install Requirements

这里有一些最常用的依赖包。如果你有其它需求，你可以上github搜索。

Here are some of the most commonly used dependency packages. If you have other requirements, you can search them on github.

pip install flask sqlalchemy graphene flask-graphql flask-sqlalchemy graphene-sqlalchemy
# MySQL
pip install mysqlclient
# PostgreSQL
pip install psycopg2-binary

运行服务器 | Run Server

首先，你需要创建一个flask app，用PyCharm很简单就能搞定。然后你应该把下面的代码添加到你的app里。我觉得你应该知道它们应该放到哪。

First, you need create a flask app. You can easily do it by PyCharm. Then please add codes below to your app. I think you know where they should be.

import graphene
from flask_graphql.graphqlview import GraphQLView

class Query(graphene.ObjectType):
    foo = graphene.String()

schema = graphene.Schema(query=Query, auto_camelcase=False)
app.add_url_rule('/graphql', view_func=GraphQLView.as_view('graphql', 
                 schema=schema, graphiql=True))

查询 | Query

如果你有公网，你可以直接在浏览器访问 http://localhost:5000/graphql 。这是GraphiQL，对新手非常友好的一个前端。但是如果你没有的话，就只能自己手动发送POST请求了。

If you have a public network, you can access http://localhost:5000/graphql in your browser. It atually is GraphiQL, friendly for new hand. But if not, you should consider POST request by yourself.

curl命令：

curl command:

curl -H "Content-Type: application/json" \
     -d '{"query": "{foo}"}' \
     http://localhost:5000/graphql

python请求：

python requests:

import requests

requests.post('http://localhost:5000/graphql', json={
    'query': '{foo}'
})

新字段 | New Fields

Hello World!

让我们在上面的Query类中创建一个"Hello World!"字段。

Let’s create a “Hello World!” field in Query class above…

class Query(graphene.ObjectType):
    hello_world = graphene.String()

    @staticmethod
    def resolve_hello_world(obj, info, **kwargs):
        return "Hello World!"

好！尝试发送一个请求吧！

Cool! Try to post a query now!

requests.post('http://localhost:5000/graphql', json={
    'query': '{hello_world}'
})

简单字段 | Simple Field

如上所述，一个简单的字段需要声明字段类型并定义一个该字段的解析器。GraphQL有很多内建类型（标量类型），像Int，Float，String，Boolean等。你可以在graphene.types里找到它们。它们的实现和Hello World差不多。

As we explained above, simple field should declare a field type and define a field resolver. GraphQL has a lot of built-in field types (Scalar Type), such as Int, Float, String, Boolean, etc. You can find all of them in graphene.types. All of them are similar to your Hello World.

你可以给自己的字段加上描述，把参数description={some_text}传递给你的类型定义即可。你可以在GraphiQL上看到你的描述。

You could also add some description to your field. Just pass description={some_text} to your type definition. You could see your descriptions on GraphiQL.

你可能还想自定义一些参数，这对于复杂查询非常有用。你应该把你的参数名和类型传递到你的字段类型定义中。参数类型和字段类型其实有一点小区别，但是现在你就姑且当它们是一样的吧。这些参数将会被graphql传递到你的解析器中。

Another point is maybe you want to add some custom args, it’s useful for our complex queries. You should pass custom args with their types to field type definition same as description. The args type has a little different with field type, but for now, you can just think they are all the same. You can get them in your resolver function.

现在让我们来创建一个简单的计算函数：add

Now, let’s create a new field to implement a simple calculate function: add

class Query(graphene.ObjectType):
    add = graphene.Int(
        description='calculate a + b then return the result.',
        a=graphene.Int(),
        b=graphene.Int())

    @staticmethod
    def resolve_add(obj, info, a=0, b=0, **kwargs):
        return a + b

然后发送一个请求。

Then post a request.

requests.post('http://localhost:5000/graphql', json={
    'query': '{add(a: 4, b: 5)}'
})

这是最后一个完整请求的示例，你现在应当明白一个正确的请求应该怎么写了。

This is the last complete query example. You should understand how to write a correct query now.

列表字段 | List Field

你可能返回一个列表，在这时你应该使用类型List。你必须同时指定列表元素的类型，并且所有元素类型必须匹配。

Maybe you need to return a list of results. At this time, you should use type List. You must pass item type to type definition. And all the item must match this type.

import random

class Query(graphene.ObjectType):
    rand = graphene.List(
        graphene.Int,
        description='get some random numbers from 0 to 100',
        count=graphene.Int(),
    )

    @staticmethod
    def resolve_rand(obj, info, count=1, **kwargs):
        return [random.randint(0, 100) for i in range(count)]

自定义（字典）字段 | Custom (Dict) Field

自定义字段是最复杂但又是最基本的一个字段。所有的自定义字段必须继承自ObjectType并且被包裹在Field类型中。

Custom field is the most complex but the most basic one. All of custom fields must be inherited from ObjectType and be wrapped into type Field.

你可以在自定义字段中声明我们上述的所有类型，包括其它自定义类型，甚至它自己。

You could declare all the fields we explained above inside a custom field, you could also declare another custom field inside it, even you could declare itself.

class CustomObject1(object):
    def __init__(self):
        super(CustomObject1, self).__init__()
        self.string = 'abc'
        self.int = 123
        self.list = ['abc', '123']

class CustomField1(graphene.ObjectType):
    # declare String field
    string_field = graphene.String()
    # declare Int field
    int_field = graphene.Int()
    # declare List field
    list_field = graphene.List(graphene.String)

    @staticmethod
    def resolve_string_field(obj, info, **kwargs):
        return obj.string

    @staticmethod
    def resolve_int_field(obj, info, **kwargs):
        return obj.int

    @staticmethod
    def resolve_list_field(obj, info, **kwargs):
        return obj.list

class CustomField2(graphene.ObjectType):
    # declare another custom field
    custom_field1 = graphene.Field(CustomField1)
    # declare itself
    custom_field2 = graphene.Field(lambda: CustomField2)

    @staticmethod
    def resolve_custom_field1(obj, info, **kwargs):
        return CustomObject1()

    @staticmethod
    def resolve_custom_field2(obj, info, **kwargs):
        return object()

上层解析器的返回值将会被传递给下层解析器的第一个参数。例如我们在resolve_custom_field1里返回了CustomObject1()，那么resolve_string_field, resolve_int_field, resolve_list_field里的obj都将是前面的CustomObject1()。如果你返回None，子解析器将不会被调用。

The parent resolver return value will be passed to the children resolvers as the first param. e.g. We return CustomObject1() at resolve_custom_field1, so all the obj in resolve_string_field, resolve_int_field, resolve_list_field equal to this CustomObject1(). If you return None, children resolvers will not be called.

顶级解析器的obj将会使None。

At root level, obj will be None.

Relay

简介 | Introduction

Relay - a JavaScript framework for building data-driven React applications

这是摘自Relay的官方描述。它和React深度结合，收集React UI 组件中的graphql查询，并将它们合并到一次请求中。这将显著的提高查询性能。

This is relay’s description comes from Relay. It combines with React depth. It will collect all the graphql definitions in React UI components, then merge them in one query. This will significantly improve query performance.

在后端，relay定义了一系列规则来协调如何对查询结果进行分页。这里是一个例子来帮助你理解它的语法。一些字段是反直觉的，务必小心。

At server site, relay define some rules to coordinate how to paginate our results. Here is a simple example to understand the syntax. Some of fields are counter-intuitive, please must be careful.

{                       #
  foo(before: $cursor,  # all the results will before this cursor
                        # (does not contain)
      after: $cursor,   # all the results will after this cursor
                        # (does not contain)
      first: $number,   # how many results will be contained in results after
                        # {after} cursor
      last: $number     # how many results will be contained in results before
                        # {before} cursor
                        #
                        # if both {before} and {after} cursor are existed, the
                        # results will be constrained in this open interval
                        # (after, before)
                        #
                        # if both {first} and {last} number are existed, the
                        # results will follow {first} only
  ) {                   #
    pageInfo {          # information of the result page
      startCursor       # the first result's cursor in this page
      endCursor         # the last result's cursor in this page
      hasPreviousPage   # if the previous page existed in this interval
      hasNextPage       # if the next page existed in this interval
                        #
                        # previous and next page will only be checked in
                        # {after} to {before} interval. example:
                        # if {after} = 10 and {first} = 10, the results will
                        # contain index from 11 to 20. but at this case, the
                        # {hasPreviousPage} flag will be set to False, because
                        # there is no previous page in interval (10, +∞). The
                        # same resone, if {first} = 10 and {last} = 10,
                        # {hasNextPage} flag will be set to False because
                        # there is no next page in interval (-∞, -10)
    }                   #
    edges {             # contains all the results
      cursor            # current cursor
      node {            # fields of this result 
        ...             #
      }                 #
    }                   #
  }                     #
}                       #

Relay字段 | Relay Field

大多数情况下，relay必须由四个级别的定义组成：Object，Node，Connection和ConnectionField。让我们看看下面的例子。

In most cases, relay must have four levels definitions: Object, Node, Connection and ConnectionField. Let’s see the simple example below.

Object级别 | Object Level

Object级别是你自己的数据类，它的一切操作完全由你决定。你也应该在其中实现其它对你有帮助的方法。

Object level is your own dataclass. It’s all decided by you. You could also define some other helper functions in this level.

class Example(object):
    def __init__(self, id, name):
        super(Example, self).__init__()
        self.id = id
        self.name = name

Node级别 | Node Level

Node级别实际上就是一个拥有特殊接口的自定义字段。你应该把你的其它字段声明在这里。

Node level is actually a custom field, but have a special interface. And you should also declare your fields types here.

from graphene import relay

class ExampleNode(graphene.ObjectType):
    id = graphene.GlobalID()
    name = graphene.String()

    @staticmethod
    def resolve_id(obj, info, **kwargs):
        return relay.Node.to_global_id('Example', obj.id)

    @staticmethod
    def resolve_name(obj, info, **kwargs):
        return obj.name

    class Meta:
        interfaces = (relay.Node,)

Connection级别 | Connection Level

Connection级别只是在relay字段和relay node之间搭了一座桥。如果你不需要做其它操作，那你创建一个空的connection就足够了。

Connection level just make a connection between relay field and relay nodes. If you don’t need other operations, you can just create a new connection type by the default function.

class ExampleConnection(relay.Connection):
    class Meta:
        node = ExampleNode

ConnectionField级别 | ConnectionField Level

ConnectionField级别给relay connection返回一个结果集。Relay提供了一个模板类，你只需要实现它的解析器接口即可。

ConnectionField level need to return a list of results for relay connection. Relay provides a template class for you, you must implement the resolver interface by yourselves.

from graphene.relay.connection import IterableConnectionField

class ExampleConnectionField(IterableConnectionField):
    @classmethod
    def resolve_connection(cls, connection_type, args, resolved):
        example_objs = [Example(i, str(i)) for i in range(100)]
        return super(ExampleConnectionField, cls).resolve_connection(
            connection_type, args, example_objs)

example_objs必须是一个Example实例的列表，并且它们将会被传递给ExampleNode的解析器作为第一个参数（我们上面已经说过了）。在基类中，这个列表将会被转换为edge列表，所以你必须调用super.resolve_connection并传递它作为第三个参数。如果你不了解其它参数是干什么的，那就把它们统统丢给基类处理吧。

The example_objs must be a list of Example instances. And items of this list will be passed to ExampleNode’s resolvers as the first param (We called it obj above). This list will be converted to edge list in the base class, so
you must call super.resolve_connection and pass this list as the third param. If you don’t understand other params, just pass them to base class.

查询Relay字段 | Query Relay Field

现在你可以把你的relay字段添加到Query中了。它和自定义字段差不多，不过你不需要再定义一遍解析器了，因为你已经在ConnectionField中定义过了。

Now you can add your relay field to your Query. It’s similar to your custom field, but you don’t need to define a resolver again, because you have defined it at ConnectionField level.

class Query(graphene.ObjectType):
    examples = ExampleConnectionField(ExampleConnection)

现在，一切工作完成，你可以试着查询一下你的第一个relay字段了。

All the works have been done. You can try to query your first relay field now.

{
  examples (first: 10, after:"YXJyYXljb25uZWN0aW9uOjM=") {
    edges {
      cursor
      node {
        id
        name
      }
    }
  }
}

这个查询的结果是数字4到13。cursor是一个包含了连接类型和列表索引的base64编码的字符串。你可以执行base64.b64decode('YXJyYXljb25uZWN0aW9uOjM=')，它的结果是'arrayconnection:3'。

The results of this query is number 4 to 13. Let you more clearer, the cursor is a base64 encoded string contains connection type and list index. You can run base64.b64decode('YXJyYXljb25uZWN0aW9uOjM='), then you can see it is actually 'arrayconnection:3'.

SQLAlchemy

SQLAlchemy Model

SQLAlchemy是一个稳定，灵活却比较复杂的ORM框架，但是我们无须了解它的内部实现，我们只需要按照推荐的方法连接数据库即可。如果你对它感兴趣，你可以去读SQLAlchemy的文档，那个文档比这个要好多了。

SQLAlchemy is a stable, flexible but complex ORM framework. But we don’t need to really understand its internal implementation. We can just use some recommended function to connect to database. If you are interested with other options, you can read SQLAlchemy docs. Its docs are better than this one.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

db_url= 'sqlite:///path/to/database/file'

engine = create_engine(db_url, convert_unicode=True, pool_recycle=3600)

session = scoped_session(sessionmaker(autocommit=True, bind=engine))

Base = declarative_base()

class ExampleTable(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String)

session().query(ExampleTable).filter(ExampleTable.id > 100).all()

SQLAlchemy Field

感谢graphene-sqlalchemy，我们可以轻松地将SQLAlchemy model转换为GraphQL字段

Thanks for the thirdparty module graphene-sqlalchemy, we can easily convert a SQLAlchemy model to a GraphQL field.

和relay字段相似，我们必须从SQLAlchemyObjectType构造一个node。但是我们不再需要创建连接类型和node自己的类型了，它们将会被自动生成。

Similar to relay field, we must construct a node from SQLAlchemyObjectType. But we don’t need to create a connection type and node’s fields. They will be generated automatically.

from graphene_sqlalchemy import SQLAlchemyObjectType

class ExampleTableNode(SQLAlchemyObjectType):
    class Meta:
        interfaces = (relay.Node,)
        model = ExampleTable

接下来你就可以通过SQLAlchemyConnectionField把它们添加到Query中了。但是这个字段不知道如何连接数据库，它将会试着在model中获取query属性，所以不要忘了把默认query赋给model。

Then you can add this node to Query through SQLAlchemyConnectionField. But this field does not know how to connect to database, it will try to find attribute query in your model as the default query, so don’t forget to assign this attribute.

from graphene_sqlalchemy import SQLAlchemyConnectionField

ExampleTable.query = session().query(ExampleTable)

class Query(graphene.ObjectType):
    examples = SQLAlchemyConnectionField(ExampleTableNode)

如果你想要动态地管理session或者动态地构造查询，包括添加查询参数，你必须继承SQLAlchemyConnectionField并重写get_query方法。

If you want to control session or build query dynamically, contains add query args, you must inherit from SQLAlchemyConnectionField and override the class method get_query.

class ExampleTableConnectionField(SQLAlchemyConnectionField):
    @classmethod
    def get_query(cls, model, info, sort=None, **args):
        query = session().query(model)
        if sort is not None:
            if isinstance(sort, str):
                query = query.order_by(sort.value)
            else:
                query = query.order_by(*(col.value for col in sort))
        return query

class Query(graphene.ObjectType):
    examples = ExampleTableConnectionField(ExampleTableNode)

现在你就可以用同样的语句（Query Relay Field）从数据库中查询数据了。

Now you can use the same query we explained above (Query Relay Field) to query data from database.