python - 如何在连接上使用石墨烯数据加载器以及如何仅选择必要的字段
问题描述
我想使用一些带有 python 石墨烯的非标准数据库来创建 GraphQL 端点。该数据库没有现有的 SQLAlchemy 方言，Django ORM 等不支持，所以我尝试使用DataLoader类来批量提取数据并最大限度地减少对数据库的查询次数。我阅读了https://docs.graphene-python.org/en/latest/execution/dataloader/。为了说明我的问题，我创建了下一个简化的即用型示例：
import graphene
from graphql.execution.executors.asyncio import AsyncioExecutor
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.graphql import GraphQLApp
from graphene.relay import Node, Connection, ConnectionField
from graphene import String, ObjectType, Field, List, NonNull
from promise import Promise
from promise.dataloader import DataLoader
class User(ObjectType):
"""User information."""
class Meta:
interfaces = (Node,)
user_id = String(description='User id.', name='userID')
user_name = String(description='User name.')
@classmethod
def get_node(cls, info, _id):
return user_loader.load(_id)
class UserConnection(Connection):
class Meta:
node = User
def get_user(_id):
return User(id=_id,
user_id=_id,
user_name='Bill Gates',
)
user_data = {x.id: x for x in [get_user(f'EEE-{x}') for x in range(8)]}
class UserLoader(DataLoader):
def batch_load_fn(self, keys):
return Promise.resolve([user_data.get(key) for key in keys])
user_loader = UserLoader()
class Query(ObjectType):
user = Field(User,
user_id=String(description='User ID', name='userID', required=True),
description='User Information')
users = List(User,
user_ids=List(NonNull(String), name='userIDs', required=True),
required=True)
all_users = ConnectionField(UserConnection)
node = Node.Field()
def resolve_user(self, info, user_id):
return user_loader.load(user_id)
def resolve_users(self, info, user_ids):
return user_loader.load_many(user_ids)
def resolve_all_users(self, info, **args):
return list(user_data.values())
routes = [
Route('/',
GraphQLApp(schema=graphene.Schema(query=Query),
executor_class=AsyncioExecutor)
),
]
app = Starlette(routes=routes)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host='127.0.0.1', port=8000)
我希望以下 GraphQL 查询能够正常工作：
fragment userFields on User {
userID
userName
}
query getUser {
user(userID: "EEE-3") {
id
...userFields
}
}
query getUsers {
users(userIDs: ["EEE-3", "EEE-4", "EEE-5"]) {
id
...userFields
}
}
query getNode {
node(id: "VXNlcjpFRUUtMw==") {
id
... on User {
...userFields
}
}
}
query getAllUsers {
allUsers(first: 2) {
edges {
node {
id
...userFields
}
cursor
}
pageInfo{
hasNextPage
hasPreviousPage
startCursor
endCursor
}
}
}
在这里，我不使用该数据库来保持示例的简单性，因为我的问题是关于如何DataLoader正确使用，或者我应该做些什么来达到我的目的。我在user_data变量中有数据并使用get_user函数访问它们，但在实际情况下，这将是对我的数据库的查询。
除了两件事之外，这里一切似乎都运行良好：
该函数
resolve_all_users直接作用于基础数据集或完整查询。虽然我希望它充当一些分页界面，根据诸如 等参数仅提取必要的first数据after。我可以在类上显式调用resolve_connection函数UserConnection，我看到它使用了一些数组切片函数。但Dataloader在示例中使用Userid 而不是索引。此外，创建索引似乎很昂贵，Dataloader因为我想获得连续间隔并且不想将所有索引作为列表传递。我应该怎么做才能只使用一个请求来解析这个字段？即使不是所有字段都需要，
DataLoader我也从数据库中提取完整的对象。User是否可以对其进行优化以仅提取 GraphQL 查询字段中所需的内容？
提前致谢。