Фильтровать по идентификатору в запросе graphene-sqlalchemy

Полную рабочую демонстрацию можно найти по адресу 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.Argument во время создания экземпляра graphene.Field, которые также должны быть названы в методе распознавателя. Однако их комбинация позволяет выполнять все виды фильтрации.

Вам необходимо указать 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()
...

Я не работал с sql-алхимией, но полагаю, вам нужно добавить интерфейс Node в свою модель, например:

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

Универсальное решение для фильтрации в графене. Прочитав тонны комментариев в Интернете, я сделал это -

Учтите, что это ваш объект, который ссылается на db.Model с именем 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()

Я надеюсь, что этот класс поможет другим. Дополнительные сопоставления преобразования типа экземпляра можно найти, если поискать и просмотреть файл graphene converter.py в Интернете.

графеновое реле, которое вы можете использовать, если вам нужны объекты по идентификатору. В этом случае идентификатор будет идентификатором узла ретрансляции, который вы можете изменить в соответствии с вашими потребностями.

Вы можете получить хороший пример по ссылке ниже:

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