Filter by id in graphene-sqlalchemy query

Question

How can you setup graphene-sqlalchemy to filter an object by id?

I'd like to run the query:

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

I would expect to get a single Marker object where markerId is 1, but I get the error "Unknown argument "markerId" on field "marker" of type "Query"."

I have two files:

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])

Thanks for your help!

Answer 1

I haven't worked with the sql-alchemy, but I suppose you have to add Node interface to your model like:

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

Answer 2

You need to specify marker_id as a query parameter when you define the query.

...
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

A versatile solution for filtering in graphene. After reading tonnes of comments online I made this -

Consider this is your Object which refers to a db.Model called Post

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

Then for the query :

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

Write this as a class in a separate file

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()

I hope this class helps others. More instance type conversion mappings can be found if you search and go through graphene converter.py file online.

Answer 4

A full working demo can be found under https://github.com/somada141/demo-graphql-sqlalchemy-falcon.

Consider the following SQLAlchemy ORM class:

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,
    )

Simply wrapped in an SQLAlchemyObjectType as such:

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

and exposed through:

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

A GraphQL query such as:

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

wil yield the response:

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

As you can see you can pass named arguments via the graphene.Argument class during the graphene.Field instantiation which also have to be named in the resolver method. The combination of the two, however, allows you to do all sorts of filtering.

Answer 5

graphene relay you can use if you want objects by ID. In this case ID will be relay node ID, which you can change as per your ID need.

You can get a godd example in below link:

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