主要内容
实现GraphQL的API查询。
- 使用 graphene 库创建GraphQL服务
- 使用Flask提供对外访问
- 使用浏览器访问GraphQL API
阅读前，你需要熟悉GraphQL。
前提设定
假设一个处理Person的应用，Person包含first_name 、last_name 和 age。graphql服务查询都在Person上执行。
# 返回一个person的所有属性
http://localhost:5000/?search={ person {firstName lastName age}}
# 返回一个person的firstName、lastName
http://localhost:5000/?search={ person {firstName lastName}}
# 返回所有person的firstName、lastName
http://localhost:5000/?search={ people {firstName lastName}}
# 返回key=1的person的所有属性
http://localhost:5000/?search={ person(key: 1) {firstName lastName age}}
为了专注在graphql使用上，数据内置在程序内。
在文件 hello_graphene.py，添加如下代码：
import collections
Person = collections.namedtuple("Person", ['first_name', 'last_name', 'age'])
data = {
1: Person("steve", "jobs", 56),
2: Person("bill", "gates", 63),
3: Person("ken", "thompson", 76),
4: Person("guido", "rossum", 63)
}
在ipython shell里进行验证
In [1]: from hello_graphene import data
...:
In [2]: data
Out[2]:
{1: Person(first_name='steve', last_name='jobs', age=56),
2: Person(first_name='bill', last_name='gates', age=63),
3: Person(first_name='ken', last_name='thompson', age=76),
4: Person(first_name='guido', last_name='rossum', age=63)}
In [3]: person = data[1]
In [4]: person.first_name
Out[4]: 'steve'
In [5]: person.last_name
Out[5]: 'jobs'
In [6]:
创建服务
安装 graphene
pip install graphene
GraphQL需要一个根类型。person的字段需要在根类型上可用。在hello_graphene.py添加如下代码：
from graphene import ObjectType, String, Int, Field
class PersonType(ObjectType):
first_name = String()
last_name = String()
age = Int()
def resolve_first_name(person, info):
return person.first_name
def resolve_last_name(person, info):
return person.last_name
def resolve_age(person, info):
return person.age
class Query(ObjectType):
person = Field(PersonType)
def resolve_person(root, info):
return data[1]
通常来说，把根类型的class命名为Query。
任意GraphQL类型必须从graphene.ObjectType 继承。GraphQL要求每个字段都提供resolver function。因此我们添加了 resolve_person for person, resolve_first_name for first_name 等resolver function。
此外需要告诉GraphQL服务根类型是Query。如下代码所示：
from graphene import Schema
schema = Schema(query=Query)
这样GraphQL的服务就设定好了。
hello_graphene.py完整的文件内容如下：
import collections
from graphene import ObjectType, String, Schema, Int, Field
Person = collections.namedtuple("Person", ['first_name', 'last_name', 'age'])
data = {
1: Person("steve", "jobs", 56),
2: Person("bill", "gates", 63),
3: Person("ken", "thompson", 76),
4: Person("guido", "rossum", 63)
}
class PersonType(ObjectType):
first_name = String()
last_name = String()
age = Int()
def resolve_first_name(person, info):
return person.first_name
def resolve_last_name(person, info):
return person.last_name
def resolve_age(person, info):
return person.age
class Query(ObjectType):
person = Field(PersonType)
def resolve_person(root, info):
return data[1]
schema = Schema(query=Query)
执行查询
从ipython shell执行GraphQL查询：
In [3]: from hello_graphene import schema
In [7]: query = '{person {firstName lastName age} }'
In [8]: result = schema.execute(query)
In [9]: result.data
Out[9]:
OrderedDict([('person',
OrderedDict([('firstName', 'steve'),
('lastName', 'jobs'),
('age', 56)]))])
返回指定字段的查询：
In [10]: query = '{person {firstName} }'
In [11]: result = schema.execute(query)
In [12]: result.data
Out[12]: OrderedDict([('person', OrderedDict([('firstName', 'steve')]))])
使用Flask提供对外查询服务：
# flask_graphql.py
import json
from flask import Flask, request
from hello_graphene import schema
app = Flask(__name__)
@app.route('/graphql')
def graphql():
query = request.args.get('query')
result = schema.execute(query)
d = json.dumps(result.data)
return '{}'.format(d)
启动flask服务
$ export FLASK_APP=flask_graphql.py
$ flask run
使用graphql查询
http://127.0.0.1:5000/graphql?query={person{firstName}}
想要查询任意person，作如下修改
person = Field(PersonType, key=Int())
同时修改resolve_person函数
def resolve_person(root, info, key):
return data[key]
上述变量key可以是任意名字，例如改为id
person = Field(PersonType, id=Int())
def resolve_person(root, info, id):
return Person.query.get(id)
使用如下URL进行查询
http://127.0.0.1:5000/graphql?query={person(id:2){firstName lastName age}}
获取person列表people
获取系统内所有person的详情，添加people在根类型上：
from graphene import List
class Query(ObjectType):
person = Field(PersonType, id=Int())
people = List(PersonType)
def resolve_person(root, info, id):
return data[id]
def resolve_people(root, info):
return data.values()
使用如下URL查询所有person
http://127.0.0.1:5000/graphql?query={ people {firstName lastName age}}
支持确省查询
目前在person上的查询必须提供id，没有id会产生异常。解决这个问题，可以提供缺省值。
person = Field(PersonType, id=Int(default_value=1))
这样，就支持无id查询，默认返回id为1的数据。
使用如下URL验证无id缺省查询。
http://127.0.0.1:5000/graphql?query={person{firstName}}
网友评论