1. Introducción

Hola a todos y bienvenidos al segundo post de 2020 de nuestro querido blog!. En el anterior vimos como crear una aplicación flask para definir un endpoint de graphql. En esta ocasión utilizaremos graphene-sqlalchemy-filter para definir y utilizar filtros complejos en nuestras consultas. Utilizamos sqlite como motor de db y testearemos nuestro código de través de unittest.

Esta librería permite definir expresiones del tipo not-equals, in o like.

Key Descripción GraphQL sufijo
eq equal
ne not equal Ne
like like Like
ilike insensitive like Ilike
is_null is null IsNull
in in In
not_in not in NotIn
lt less than Lt
lte less than or equal Lte
gt greater than Gt
gte greater than or equal Gte
range in range Range
contains contains (PostgreSQL array) Contains
contained_by contained_by (PostgreSQL array) ContainedBy
overlap overlap (PostgreSQL array) Overlap

Se definen a través del argumento filters

    query{
       user(filters: {useridIn: [1, 2]}){
          edges{
             node{
                userid
                name
                surname
                age
              }
            }
        }
    }

Tecnologías empleadas:

  • Python 3.6
  • Flask 1.1.1
  • Graphene 2.1.8
  • Graphene SQLAlchemy Filter 1.10.2
  • SQLAlchemy 1.3.13
  • SQLite

Os podéis descargar el código de mi GitHub que se encuentra aquí.

1. Creación virtualenv

Creamos el virtualenv e instalamos las dependencias necesarias

$ virtualenv -p python3 venv
$ source venv/bin/activate
$ pip install -r requirements.txt

flask
flask-graphql
flask-migrate
flask-sqlalchemy
graphene
graphene-sqlalchemy
graphene-sqlalchemy-filter

2. Fuentes

Creamos una instancia de la clase Flask. Además establecemos la configuración para acceder a la base de datos a través de la variable de entorno SQLALCHEMY_DATABASE_URI

from flask import Flask

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tests/sqlite.db'

Creamos el modelo asociado a la tabla user

from flask_sqlalchemy import SQLAlchemy

from app import app

db = SQLAlchemy(app)


class UserModel(db.Model):
    __tablename__ = 'user'

    userid = db.Column(db.Integer, primary_key=True)

    name = db.Column(db.String(256))

    surname = db.Column(db.String(256))

    age = db.Column(db.Integer)

    def __repr__(self):
        return '<User {} {} {} {}>'.format(self.id, self.name, self.surname, self.age)

Aquí viene una de las novedades con respecto al post anterior!. Creamos una clase que extienda de FilterSet . Nos permitirá definir los filtros que podemos aplicar a cada uno de los campos de nuestro modelo. En esta ocasión asignaremos todas las operaciones permitidas a todos los campos del modelo.

from graphene_sqlalchemy_filter import FilterSet

from model import UserModel

ALL_OPERATIONS = ['eq', 'ne', 'like', 'ilike', 'is_null', 'in', 'not_in', 'lt', 'lte', 'gt', 'gte', 'range']


class UserFilter(FilterSet):
    class Meta:
        model = UserModel
        fields = {
            'userid': ALL_OPERATIONS,
            'name': ALL_OPERATIONS,
            'surname': ALL_OPERATIONS,
            'age': ALL_OPERATIONS,
        }

Creamos el schema para GraphQL. Se instancia un objeto de la clase graphene.Schema que define las consultas que se podrán realizar. La clase FilterableConnectionField será la encargada de manejar los filtros que se ejecuten en las consultas. Además se encargará de componer las querys que se ejecutarán sobre el modelo UserModel a partir de las consultar realizadas.

import graphene
from graphene_sqlalchemy import SQLAlchemyObjectType
from graphene_sqlalchemy_filter import FilterableConnectionField

from filter import UserFilter
from model import UserModel


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


class Query(graphene.ObjectType):
    node = graphene.relay.Node.Field()
    user = FilterableConnectionField(connection=User, filters=UserFilter(), sort=User.sort_argument())


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

Creación de la vista a través del schema creado

from flask_graphql import GraphQLView

from app import app
from schema import schema

app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view(
        'graphql',
        schema=schema,
        graphiql=True
    )
)


if __name__ == '__main__':
    app.run()

3. Testeando la aplicación

Para testear la aplicación utilizamos unittest. Creamos la base de datos a partir de nuestro modelo

from model import db
db.create_all()

Insertamos usuarios de prueba

@staticmethod
def _insert_users():
   db.session.add(UserModel(userid=1, name='Jorge', surname='Hernandez', age=32))
   db.session.add(UserModel(userid=2, name='Jose', surname='Hernandez', age=32))
   db.session.commit()

Definimos el contexto de la aplicación

AppContext(app).push()

Y creamos un cliente sobre el que realizar las consultas

self.client = Client(schema)

Se muestra todo el fichero test_user.py 

import json
import os
import unittest

from flask.ctx import AppContext
from graphene.test import Client

from app import app
from model import db, UserModel
from schema import schema


class UserTest(unittest.TestCase):

    FILTER_IN = """
        query{
          user(filters: {useridIn: [1, 2]}){
            edges{
              node{
                userid
                name
                surname
                age
              }
            }
          }
        }
    """

    FILTER_NE = """
        query{
          user(filters: {useridNe: 1}){
            edges{
              node{
                userid
                name
                surname
                age
              }
            }
          }
        }
    """

    FILTER_LIKE = """
        query{
          user(filters: {nameLike: "%os%"}){
            edges{
              node{
                userid
                name
                surname
                age
              }
            }
          }
        }
    """

    @staticmethod
    def _create_database_mock():
        db.drop_all()
        db.create_all()

    @staticmethod
    def _insert_users():
        db.session.add(UserModel(userid=1, name='Jorge', surname='Hernandez', age=32))
        db.session.add(UserModel(userid=2, name='Jose', surname='Hernandez', age=32))
        db.session.commit()

    @classmethod
    def setUpClass(cls) -> None:
        os.environ['FLASK_ENV'] = 'test'
        AppContext(app).push()
        UserTest._create_database_mock()
        UserTest._insert_users()

    def setUp(self):
        self.client = Client(schema)

    def test_should_be_not_none_client(self):
        self.assertIsNotNone(self.client)

    def test_should_validate_in_operator(self):
        self.assertEqual(
            {"data": {"user": {"edges": [{"node": {"userid": "1", "name": "Jorge", "surname": "Hernandez", "age": 32}}, {"node": {"userid": "2", "name": "Jose", "surname": "Hernandez", "age": 32}}]}}},
            self.client.execute(self.FILTER_IN))

    def test_should_validate_not_equal_operator(self):
        self.assertEqual(
            {"data": {"user": {"edges": [{"node": {"userid": "2", "name": "Jose", "surname": "Hernandez", "age": 32}}]}}},
            self.client.execute(self.FILTER_NE))

    def test_should_validate_like_operator(self):
        self.assertEqual(
            {"data": {"user": {"edges": [{"node": {"userid": "2", "name": "Jose", "surname": "Hernandez", "age": 32}}]}}},
            self.client.execute(self.FILTER_LIKE))

if __name__ == '__main__':
    unittest.main()

4. Referencias