Flexible arguments in SQLAlchemyConnectionField and custom query to filter against arguments

[kororo]

Hi maintainers,

I have this situation that I need more args to filter the SQLAlchemyConnectionField. Currently from graphene, it gives us (before, after, first, last). My objective is to add additional args for each Model included.

Here be dragons.

Models and theirs attributes:
Job: name, blocks
Block: name, job
Note: 1 Job has 0 or more Blocks

Arguments:
jobs: name, before, after, first, last
blocks: before, after,first,last

{
  jobs(name: "test") {
    edges {
      node {
        id
        name
        blocks {
          edges {
            node {
              id
            }
          }
        }
      }
    }
  }
}

Notes:

• NOTE1: this is very important to get this working, maybe need some relationship checking in the code. Without the dynamic relationship, the relationship is not configurable as Query object.
• NOTE2: this is the base model for the GQL schema, I am thinking to have two filter functions to help on the NOTE3 and NOTE4
• NOTE3: the filter/hook to add more arguments for example shown in NOTE5
• NOTE4: the filter/hook to process the arguments for example shown in NOTE6
• NOTE5: adding name as additional args
• NOTE6: process the name to the existing query in relationship
• NOTE7: I subclassing your existing class, I hope this proposal is accepted and incorporated into the master.
• NOTE8: so glad you did this maintainer, this is the way for me to intercept the class instantiation
• NOTE9: the way to set the field connection now

class Job(Model):
    name = sa.Column(sa.String)
    # NOTE1
    blocks = sa.relationship('block', back_populates='job', lazy='dynamic')

class Block(Model):
    name = sa.Column(sa.String)
    # NOTE1
    job = sa.relationship('job', back_populates='blocks', lazy='dynamic')

# NOTE2
class BaseObjectType(SQLAlchemyObjectType):
    class Meta:
        abstract = True

    # NOTE3
    @classmethod
    def update_connection_args(cls, **kwargs):
        return kwargs

    # NOTE4
    @classmethod
    def process_args(cls, query, root, connection, args: dict, **kwargs):
        return query

class JobObjectType(BaseObjectType):
    class Meta:
        model = Job
        interfaces = (graphene.relay.Node,)

    @classmethod
    def update_connection_args(cls, **kwargs):
        kwargs = super(JobObjectType, cls).update_connection_args(**kwargs)
        # NOTE5
        kwargs.setdefault('name', String(required=False))
        return kwargs

    @classmethod
    def process_args(cls, query, root, connection, args: dict, **kwargs):
        # NOTE6
        name = args.get('name')
        if name:
            query = query.filter(Job.name == name)
        return query

class BlockObjectType(BaseObjectType):
    class Meta:
        model = Block
        interfaces = (graphene.relay.Node,)

class JobConnection(relay.Connection):
    class Meta:
        node = JobObjectType

class BlockConnection(relay.Connection):
    class Meta:
        node = BlockObjectType

# NOTE7
class FlexibleSQLAlchemyConnectionField(SQLAlchemyConnectionField):
    def __init__(self, model_type, *args, **kwargs):
        update_connection_args = getattr(model_type._meta.node, 'update_connection_args', None)
        if update_connection_args:
            kwargs = update_connection_args(**kwargs)
        super(FlexibleSQLAlchemyConnectionField, self).__init__(model_type, *args, **kwargs)

    @classmethod
    def connection_resolver(cls, resolver, connection, model, root, info, **args):
        iterable = resolver(root, info, **args)
        if iterable is None:
            iterable = cls.get_query(model, info, **args)

        process_args = getattr(connection._meta.node, 'process_args', None)
        if process_args:
            iterable = process_args(query=iterable, root=root, connection=connection, args=args)

        if isinstance(iterable, orm.Query):
            _len = iterable.count()
        else:
            _len = len(iterable)

        connection = connection_from_list_slice(
            iterable,
            args,
            slice_start=0,
            list_length=_len,
            list_slice_length=_len,
            connection_type=connection,
            pageinfo_type=PageInfo,
            edge_type=connection.Edge,
        )

        connection.iterable = iterable
        connection.length = _len
        return connection


# NOTE8
def register_connection_field(_type):
    return FlexibleSQLAlchemyConnectionField(_type)

registerConnectionFieldFactory(register_connection_field)


class Query(graphene.ObjectType):
    node = graphene.relay.Node.Field()
    # NOTE9
    blocks = FlexibleSQLAlchemyConnectionField(BlockConnection)
    jobs = FlexibleSQLAlchemyConnectionField(JobConnection)

schema = graphene.Schema(
    query=Query,
    types=[JobObjectType, BlockObjectType]
)

Apologies for long explanation, I really hope this illustrates the situation I am facing and the proposal from me to add two additional filters/hooks into the SQLAlchemyConnectionField. I am sure this proposal will be greatly accepted by community, since from my perspective this is immediate update that I require to do after installing your module.

I am happy to make the PR changes including the test to the master, just let me know whether the hook/filter naming convention is good, the arguments in the def is good and also the way that I approach to my problem is acceptable. I am open to any discussion.

Thanks!!

[wakemaster39]

I have been implementing something similar myself and have one suggested change to the implementation here. The code in connection_resolver can be moved to resolve_connection and have a pretty minimal impact on all code below it.

It made implementing this externally a lot cleaner for me

    @classmethod
    def resolve_connection(cls, connection_type, model, info, args, resolved):
        if resolved is None:
            resolved = cls.get_query(model, info, **args)
        process_args = getattr(connection_type._meta.node, 'process_args', None)
        if process_args:
            resolved = process_args(query=resolved, info=info, connection=connection_type, args=args)

        return super().resolve_connection(connection_type, model, info, args, resolved)

[matutter]

A builtin way to support declaring extra inputs for the SQLAlchemyConnectionField would be a huge improvement. This is the only part of graphene-sqlalchemy where I actually have to extend something to fit my use-case.

This is a much less complete solution, but just for an idea about what people are doing to work around it. But does what I need it to 99% of the time.

class RoleInput(graphene.InputObjectType):
  role = String()

class FilteredConnectionField(SQLAlchemyConnectionField):

  def __init__(self, type, input_type, *args, **kwargs):
    fields = {name: field.type() for name, field in input_type._meta.fields.items()}
    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

class Query(graphene.ObjectType):
  users = FilteredConnectionField(Users, RoleInput)

[cglacet]

What about graphene-sqlalchemy-filter? That seems to achieve what you need, no?

[ShivanshJ]

As a developer, a feedback - the learning curve of graphene-sqlalchemy increases so much when so many new libraries ,and then their evermore classes, involved.
I am facing a tough time understanding connections, including authentications etc. Using another filtering library seems like an overkill to me.
Can't simple filters be included like in the django-graphene library ?

[ShivanshJ]

FilteredConnectionField

How do I use this on a Schema Object which uses relay for its interface, like this one -

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

[ShivanshJ]

class Query(graphene.ObjectType):
users = FilteredConnectionField(Users, RoleInput)

Thanks to you I extended it into something more versatile for filter fields in Flask / graphQL / sqlalchemy :

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

Then for the query :

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

and here PostObject is simply :

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

I hope this class helps others.
More instance type conversion mappings can be found in graphene converters.py File.

[shota]

I finally found the way to do this in the right way. Simply add graphene.Argument to kwargs of SQLAlchemyConnectionField like below.

class User(SQLAlchemyObjectType):
  class Meta:
    model = UserModel
    interfaces = (relay.Node, )

...

class Query(graphene.ObjectType):
  users = SQLAlchemyConnectionField(User.connection, status=graphene.Argument(graphene.String))
  def resolve_users(self, info=None, sort=None, status=None):
    return fetch_users(status=status)

We can accept the status argument in Query.

{
  users(status:"active") {
    edges {
      node {
        id
        name
      }
    }
  }

You can also accept graphene.List(graphene.String) for list of string. That would be nice for querying status. Of course, many other types of argument which graphene supports would work.

Hope this helps.

[jqhoogland]

Following up on @shota's solution, if you consistently need the same arguments, consider including a default among **kwargs in SQLAlchemyConnectionField.__init__.

class MySQLAlchemyConnectionField(SQLAlchemyConnectionField):
    def __init__(self, *args, **kwargs):
        super(*args, status=graphene.Argument(graphene.String), **kwargs)
       
class Query(graphene.ObjectType):
    users = MySQLAlchemyConnectionField(User.connection)
    def resolve_users(self, info=None, sort=None, status=None):
        return fetch_users(status=status)

[erikwrede]

I am closing all old issues&PRs related to filtering. The discussion will continue in #347 (WIP). A proposal for the implementation of filters is currently being worked on and will be posted there once it is ready.

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related topics referencing this issue.