在石墨烯-sqlcellhe 查询中按 id 过滤

Question

如何设置石墨烯-sqlchemy以按id过滤对象？

我想运行查询：

{
marker(markerId: 1) {
markerId
title
}
}

我希望得到一个标记为 1 的标记对象，但我在类型为"查询"的字段"标记"上收到错误"未知参数"markerId"。

我有两个文件：

schema.py

import graphene
from graphene_sqlalchemy import SQLAlchemyObjectType
from model import db_session, Marker as MarkerModel
class Marker(SQLAlchemyObjectType):
class Meta:
model = MarkerModel
class Query(graphene.ObjectType):
marker = graphene.Field(Marker)
markers = graphene.List(Marker)
def resolve_markers(self, args, context, info):
return db_session.query(MarkerModel).all()
def resolve_marker(self, args, context, info):
return db_session.query(MarkerModel).first()
schema = graphene.Schema(query=Query)

model.py

import sqlalchemy as db
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from instance.config import settings
engine = db.create_engine(settings["SQLALCHEMY_DATABASE_URI"])
sm = sessionmaker(bind=engine)
db_session = scoped_session(sm)
Base = declarative_base()
Base.query = db_session.query_property()

class Marker(Base):
__tablename__ = "marker"
marker_id = db.Column(db.Integer, primary_key=True)
latitude = db.Column(db.Float)
longitude = db.Column(db.Float)
title = db.Column(db.String(100))
blurb = db.Column(db.String(65535))
def __repr__(self):
return "<Marker %d: %s>".format([self.marker_id, self.title])

感谢您的帮助！

Answer 1

完整的工作演示可以在 https://github.com/somada141/demo-graphql-sqlalchemy-falcon 下找到。

请考虑以下 SQLAlchemy ORM 类：

class Author(Base, OrmBaseMixin):
__tablename__ = "authors"
author_id = sqlalchemy.Column(
sqlalchemy.types.Integer(),
primary_key=True,
)
name_first = sqlalchemy.Column(
sqlalchemy.types.Unicode(length=80),
nullable=False,
)
name_last = sqlalchemy.Column(
sqlalchemy.types.Unicode(length=80),
nullable=False,
)

简单地包裹在这样的SQLAlchemyObjectType中：

class TypeAuthor(SQLAlchemyObjectType):
class Meta:
model = Author

并通过以下方式公开：

author = graphene.Field(
TypeAuthor,
author_id=graphene.Argument(type=graphene.Int, required=False),
name_first=graphene.Argument(type=graphene.String, required=False),
name_last=graphene.Argument(type=graphene.String, required=False),
)
@staticmethod
def resolve_author(
args,
info,
author_id: Union[int, None] = None,
name_first: Union[str, None] = None,
name_last: Union[str, None] = None,
):
query = TypeAuthor.get_query(info=info)
if author_id:
query = query.filter(Author.author_id == author_id)
if name_first:
query = query.filter(Author.name_first == name_first)
if name_last:
query = query.filter(Author.name_last == name_last)
author = query.first()
return author

一个 GraphQL 查询，例如：

query GetAuthor{
author(authorId: 1) {
nameFirst
}
}

将产生响应：

{
"data": {
"author": {
"nameFirst": "Robert"
}
}
}

如您所见，您可以在graphene.Field实例化期间通过graphene.Argument类传递命名参数，这些参数也必须在解析器方法中命名。但是，两者的结合允许您进行各种过滤。

Answer 2

定义查询时，需要将marker_id指定为查询参数。

...
class Query(graphene.ObjectType):
marker = graphene.Field(Marker, marker_id=graphene.String())
markers = graphene.List(Marker)
def resolve_markers(self, args, context, info):
return db_session.query(MarkerModel).all()
def resolve_marker(self, args, context, info, marker_id):
return db_session.query(MarkerModel).filter(MarkerModel.id == marker_id).first()
...

Answer 3

我没有使用过sql-alchemy，但我想你必须将Node接口添加到你的模型中，如下所示：

class Marker(SQLAlchemyObjectType):
class Meta:
model = MarkerModel
interfaces = (Node,)

Answer 4

用于过滤石墨烯的多功能解决方案。在网上阅读了大量评论后，我做了这个 -

考虑这是您的对象，它引用了一个数据库。称为Post的模型

class PostObject(SQLAlchemyObjectType):
class Meta:
model = Post
interfaces = (graphene.relay.Node, )

然后对于查询：

class Query(graphene.ObjectType):
node = graphene.relay.Node.Field()
all_posts = FilteredConnectionField(PostObject)

将其作为类写入单独的文件中

import graphene
import sqlalchemy
from graphene_sqlalchemy import SQLAlchemyConnectionField
class FilteredConnectionField(SQLAlchemyConnectionField):
def __init__(self, type, *args, **kwargs):
fields = {}
columns = input_type._meta.model.__table__.c
for col in columns:
fields[col.name] = self.sql_graphene_type_mapper(col.type)
kwargs.update(fields)
super().__init__(type, *args, **kwargs)
@classmethod
def get_query(cls, model, info, sort=None, **args):
query = super().get_query(model, info, sort=sort, **args)
omitted = ('first', 'last', 'hasPreviousPage', 'hasNextPage', 'startCursor', 'endCursor')
for name, val in args.items():
if name in omitted: continue
col = getattr(model, name, None)
if col:
query = query.filter(col == val)
return query
def sql_graphene_type_mapper(self, col_type):
if isinstance(col_type, sqlalchemy.Integer): return graphene.Int()
elif isinstance(col_type, sqlalchemy.Boolean): return graphene.Boolean()
elif isinstance(col_type, sqlalchemy.DateTime): return graphene.types.DateTime()
elif isinstance(col_type, (sqlalchemy.FLOAT, sqlalchemy.BIGINT, sqlalchemy.NUMERIC )): return graphene.Float()
else:
return graphene.String()

我希望这门课能帮助别人。 如果您在线搜索并浏览石墨烯converter.py文件，可以找到更多实例类型转换映射。

Answer 5

石墨烯继电器，如果你想要按 ID 排列的对象，你可以使用。在这种情况下，ID 将是中继节点 ID，您可以根据需要进行更改。

您可以在下面的链接中获得一个 godd 示例：

https://github.com/alexisrolland/flask-graphene-sqlalchemy