GraphQL + Django:使用原始 PostgreSQL 查询解析查询
GraphQL + Django: resolve queries using raw PostgreSQL query
当使用外部数据库从多个 table 中获取数据时,将 GraphQL 与 Django 结合使用的最佳方式是什么(即,创建 Django 模型来表示数据不对应到我数据库中的单个 table)?
我的方法是暂时放弃使用 Django 模型,因为我认为我还没有完全理解它们。 (我对 Django 和 GraphQL 都是全新的。)我已经建立了一个简单的项目,其中包含一个连接了外部 Postgres 数据库的应用程序。我遵循了 Graphene Django tutorial 的所有设置,然后当我意识到我创建的模型是几个 table 的混合物时遇到了障碍。
我有一个查询发送回映射到我模型中字段的正确列,但我不知道如何使它成为一个动态连接,这样当我的 API 被点击时,它查询我的数据库并将行映射到我在 Django 中定义的模型架构。
从那以后,我的方法一直是避免使用模型并使用 Steven Luscher 的演讲中演示的更简单的方法:Zero to GraphQL in 30 Minutes。
TLDR;
The goal is to be able to hit my GraphQL endpoint, use a cursor object from my django.db.connection to get a list of dictionaries that should resolve to a GraphQLList of OrderItemTypes (see below).
问题是当我通过查询访问以下端点时,每个值都为空值:
localhost:8000/api?query={orderItems{date,uuid,orderId}}
returns:
{ "data":{ "orderItems":[ {"date":null, "uuid":null, "orderId":null }, ... ] } }
project/main/app/schema.py
import graphene
from django.db import connection
class OrderItemType(graphene.ObjectType):
date = graphene.core.types.custom_scalars.DateTime()
order_id = graphene.ID()
uuid = graphene.String()
class QueryType(graphene.ObjectType):
name = 'Query'
order_items = graphene.List(OrderItemType)
def resolve_order_items(root, args, info):
data = get_order_items()
# data prints out properly in my terminal
print data
# data does not resolve properly
return data
def get_db_dicts(sql, args=None):
cursor = connection.cursor()
cursor.execute(sql, args)
columns = [col[0] for col in cursor.description]
data = [
dict(zip(columns, row))
for row in cursor.fetchall() ]
cursor.close()
return data
def get_order_items():
return get_db_dicts("""
SELECT j.created_dt AS date, j.order_id, j.uuid
FROM job AS j
LIMIT 3;
""")
在我的终端中,我从 QueryType 的 resolve 方法打印,我可以看到数据成功地从我的 Postgres 连接返回。然而,GraphQL 给了我空值,所以它必须在 resolve 方法中一些映射被搞砸了。
[ { 'uuid': u'7584aac3-ab39-4a56-9c78-e3bb1e02dfc1', 'order_id': 25624320, 'date': datetime.datetime(2016, 1, 30, 16, 39, 40, 573400, tzinfo=<UTC>) }, ... ]
如何将我的数据正确映射到我在 OrderItemType 中定义的字段?
这里还有一些参考资料:
project/main/schema.py
import graphene
from project.app.schema import QueryType AppQuery
class Query(AppQuery):
pass
schema = graphene.Schema(
query=Query, name='Pathfinder Schema'
)
file tree
|-- project
|-- manage.py
|-- main
|-- app
|-- models.py
|-- schema.py
|-- schema.py
|-- settings.py
|-- urls.py
这是临时解决方法,但我希望有更清洁的方法来处理 snake_cased 字段名。
project/main/app/schema.py
from graphene import (
ObjectType, ID, String, Int, Float, List
)
from graphene.core.types.custom_scalars import DateTime
from django.db import connection
''' Generic resolver to get the field_name from self's _root '''
def rslv(self, args, info):
return self.get(info.field_name)
class OrderItemType(ObjectType):
date = DateTime(resolver=rslv)
order_id = ID()
uuid = String(resolver=rslv)
place_id = ID()
''' Special resolvers for camel_cased field_names '''
def resolve_order_id(self, args, info):
return self.get('order_id')
def resolve_place_id(self, args, info):
return self.get('place_id')
class QueryType(ObjectType):
name = 'Query'
order_items = List(OrderItemType)
def resolve_order_items(root, args, info):
return get_order_items()
GraphQL Python/Graphene 上的默认解析器尝试使用 getattr 在根对象中解析给定的 field_name。
因此,例如,名为 order_items
的字段的默认解析器将类似于:
def resolver(root, args, context, info):
return getattr(root, 'order_items', None)
知道,当在 dict
中执行 getattr
时,结果将是 None
(要访问字典项,您将不得不使用 __getitem__
/ dict[key]
).
因此,解决您的问题就像将内容从 dicts
更改为将内容存储到 namedtuples
一样简单。
import graphene
from django.db import connection
from collections import namedtuple
class OrderItemType(graphene.ObjectType):
date = graphene.core.types.custom_scalars.DateTime()
order_id = graphene.ID()
uuid = graphene.String()
class QueryType(graphene.ObjectType):
class Meta:
type_name = 'Query' # This will be name in graphene 1.0
order_items = graphene.List(OrderItemType)
def resolve_order_items(root, args, info):
return get_order_items()
def get_db_rows(sql, args=None):
cursor = connection.cursor()
cursor.execute(sql, args)
columns = [col[0] for col in cursor.description]
RowType = namedtuple('Row', columns)
data = [
RowType(*row) # Edited by John suggestion fix
for row in cursor.fetchall() ]
cursor.close()
return data
def get_order_items():
return get_db_rows("""
SELECT j.created_dt AS date, j.order_id, j.uuid
FROM job AS j
LIMIT 3;
""")
希望对您有所帮助!
也可以只更改 graphene.object 的默认解析器。
我相信重组后以下内容会奏效。
from graphene.types.resolver import dict_resolver
class OrderItemType(ObjectType):
class Meta:
default_resolver = dict_resolver
date = DateTime()
order_id = ID()
uuid = String()
place_id = ID()
即使这个方法不能直接适用于上面的问题,这个问题是我在研究如何做到这一点时发现的。
dict_resolver
当使用外部数据库从多个 table 中获取数据时,将 GraphQL 与 Django 结合使用的最佳方式是什么(即,创建 Django 模型来表示数据不对应到我数据库中的单个 table)?
我的方法是暂时放弃使用 Django 模型,因为我认为我还没有完全理解它们。 (我对 Django 和 GraphQL 都是全新的。)我已经建立了一个简单的项目,其中包含一个连接了外部 Postgres 数据库的应用程序。我遵循了 Graphene Django tutorial 的所有设置,然后当我意识到我创建的模型是几个 table 的混合物时遇到了障碍。
我有一个查询发送回映射到我模型中字段的正确列,但我不知道如何使它成为一个动态连接,这样当我的 API 被点击时,它查询我的数据库并将行映射到我在 Django 中定义的模型架构。
从那以后,我的方法一直是避免使用模型并使用 Steven Luscher 的演讲中演示的更简单的方法:Zero to GraphQL in 30 Minutes。
TLDR;
The goal is to be able to hit my GraphQL endpoint, use a cursor object from my django.db.connection to get a list of dictionaries that should resolve to a GraphQLList of OrderItemTypes (see below).
问题是当我通过查询访问以下端点时,每个值都为空值:
localhost:8000/api?query={orderItems{date,uuid,orderId}}
returns:
{ "data":{ "orderItems":[ {"date":null, "uuid":null, "orderId":null }, ... ] } }
project/main/app/schema.py
import graphene
from django.db import connection
class OrderItemType(graphene.ObjectType):
date = graphene.core.types.custom_scalars.DateTime()
order_id = graphene.ID()
uuid = graphene.String()
class QueryType(graphene.ObjectType):
name = 'Query'
order_items = graphene.List(OrderItemType)
def resolve_order_items(root, args, info):
data = get_order_items()
# data prints out properly in my terminal
print data
# data does not resolve properly
return data
def get_db_dicts(sql, args=None):
cursor = connection.cursor()
cursor.execute(sql, args)
columns = [col[0] for col in cursor.description]
data = [
dict(zip(columns, row))
for row in cursor.fetchall() ]
cursor.close()
return data
def get_order_items():
return get_db_dicts("""
SELECT j.created_dt AS date, j.order_id, j.uuid
FROM job AS j
LIMIT 3;
""")
在我的终端中,我从 QueryType 的 resolve 方法打印,我可以看到数据成功地从我的 Postgres 连接返回。然而,GraphQL 给了我空值,所以它必须在 resolve 方法中一些映射被搞砸了。
[ { 'uuid': u'7584aac3-ab39-4a56-9c78-e3bb1e02dfc1', 'order_id': 25624320, 'date': datetime.datetime(2016, 1, 30, 16, 39, 40, 573400, tzinfo=<UTC>) }, ... ]
如何将我的数据正确映射到我在 OrderItemType 中定义的字段?
这里还有一些参考资料:
project/main/schema.py
import graphene
from project.app.schema import QueryType AppQuery
class Query(AppQuery):
pass
schema = graphene.Schema(
query=Query, name='Pathfinder Schema'
)
file tree
|-- project
|-- manage.py
|-- main
|-- app
|-- models.py
|-- schema.py
|-- schema.py
|-- settings.py
|-- urls.py
这是临时解决方法,但我希望有更清洁的方法来处理 snake_cased 字段名。
project/main/app/schema.py
from graphene import (
ObjectType, ID, String, Int, Float, List
)
from graphene.core.types.custom_scalars import DateTime
from django.db import connection
''' Generic resolver to get the field_name from self's _root '''
def rslv(self, args, info):
return self.get(info.field_name)
class OrderItemType(ObjectType):
date = DateTime(resolver=rslv)
order_id = ID()
uuid = String(resolver=rslv)
place_id = ID()
''' Special resolvers for camel_cased field_names '''
def resolve_order_id(self, args, info):
return self.get('order_id')
def resolve_place_id(self, args, info):
return self.get('place_id')
class QueryType(ObjectType):
name = 'Query'
order_items = List(OrderItemType)
def resolve_order_items(root, args, info):
return get_order_items()
GraphQL Python/Graphene 上的默认解析器尝试使用 getattr 在根对象中解析给定的 field_name。
因此,例如,名为 order_items
的字段的默认解析器将类似于:
def resolver(root, args, context, info):
return getattr(root, 'order_items', None)
知道,当在 dict
中执行 getattr
时,结果将是 None
(要访问字典项,您将不得不使用 __getitem__
/ dict[key]
).
因此,解决您的问题就像将内容从 dicts
更改为将内容存储到 namedtuples
一样简单。
import graphene
from django.db import connection
from collections import namedtuple
class OrderItemType(graphene.ObjectType):
date = graphene.core.types.custom_scalars.DateTime()
order_id = graphene.ID()
uuid = graphene.String()
class QueryType(graphene.ObjectType):
class Meta:
type_name = 'Query' # This will be name in graphene 1.0
order_items = graphene.List(OrderItemType)
def resolve_order_items(root, args, info):
return get_order_items()
def get_db_rows(sql, args=None):
cursor = connection.cursor()
cursor.execute(sql, args)
columns = [col[0] for col in cursor.description]
RowType = namedtuple('Row', columns)
data = [
RowType(*row) # Edited by John suggestion fix
for row in cursor.fetchall() ]
cursor.close()
return data
def get_order_items():
return get_db_rows("""
SELECT j.created_dt AS date, j.order_id, j.uuid
FROM job AS j
LIMIT 3;
""")
希望对您有所帮助!
也可以只更改 graphene.object 的默认解析器。
我相信重组后以下内容会奏效。
from graphene.types.resolver import dict_resolver
class OrderItemType(ObjectType):
class Meta:
default_resolver = dict_resolver
date = DateTime()
order_id = ID()
uuid = String()
place_id = ID()
即使这个方法不能直接适用于上面的问题,这个问题是我在研究如何做到这一点时发现的。
dict_resolver