sqlalchemy-json-querybuilder 1.1.8

Querybuilder to use SqlAlchemy ORM by feeding JSON/object as input

Sqlalchemy JSON Querybuilder

It introduces a middleware between your application and Sqlalchemy ORM. So input to ORM can be provided in the form JSON/Objects.

Installation

pip install sqlalchemy-json-querybuilder

Operators

Following operators are supported -

equals, notequals, lt, lte, gt, gte, like, ilike, startswith, istartswith, endswith, iendswith, contains, icontains, in, notin, isnull, isnotnull, any, has

Note - i stands for case insensitive.

• equals

filter_by = [dict(field_name='User.name', field_value='ed', operator='equals')]

is translated to

query.filter(User.name == 'ed')

• notequals

filter_by = [dict(field_name='User.name', field_value='ed', operator='notequals')]

is translated to

query.filter(User.name != 'ed')

• lt

filter_by = [dict(field_name='User.age', field_value=18, operator='lt')]

is translated to

query.filter(User.name < 18)

• lte

filter_by = [dict(field_name='User.age', field_value=18, operator='lte')]

is translated to

query.filter(User.name <= 18)

• gt

filter_by = [dict(field_name='User.age', field_value=18, operator='gt')]

is translated to

query.filter(User.name > 18)

• gte

filter_by = [dict(field_name='User.age', field_value=18, operator='gte')]

is translated to

query.filter(User.name >= 18)

• in

filter_by = [dict(field_name='User.name', field_value=['ed', 'wendy', 'jack'], operator='in')]

is translated to

query.filter(User.name.in_(['ed', 'wendy', 'jack']))

• notin

filter_by = [dict(field_name='User.name', field_value=['ed', 'wendy', 'jack'], operator='notin')]

is translated to

query.filter(~User.name.in_(['ed', 'wendy', 'jack']))

• isnull

filter_by = [dict(field_name='User.name', field_value=null, operator='isnull')]

is translated to

query.filter(User.name == None)

# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))

• isnotnull

filter_by = [dict(field_name='User.name', field_value=null, operator='isnotnull')]

is translated to

query.filter(User.name != None)

# alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))

• contains

filter_by = [dict(field_name='User.name', field_value='ed', operator='contains')]

is translated to

query.filter(User.name.like('%ed%'))

• startswith

filter_by = [dict(field_name='User.name', field_value='ed', operator='startswith')]

is translated to

query.filter(User.name.like('ed%'))

• endswith

filter_by = [dict(field_name='User.name', field_value='ed', operator='endswith')]

is translated to

query.filter(User.name.like('%ed'))

• any

filter_by = [{
    'field_name': 'User.addresses',
    'operator': 'any',
    'field_value': {
        'field_name': 'Address.email_address',
        'operator': 'equals',
        'field_value': 'bar'
    }
}]

is translated to

query.filter(User.addresses.any(Address.email_address == 'bar'))

# also takes keyword arguments:
query.filter(User.addresses.any(email_address='bar'))

• has

filter_by = [{
    'field_name': 'Address.user',
    'operator': 'has',
    'field_value': {
        'field_name': 'User.name',
        'operator': 'equals',
        'field_value': 'bar'
    }
}]

is translated to

query.filter(Address.user.has(name='ed'))

Usage

#-------------- Creating connection & session ---------------#

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

Base = declarative_base()
con_url = 'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'.format(
    username='root', password='', host='localhost', port=3306, database='test'
)
engine = create_engine(con_url, pool_recycle=3600)

# Set up the session
session_maker = sessionmaker(bind=engine, autoflush=True, autocommit=False, expire_on_commit=True)
session = scoped_session(session_maker)

#-------------- Models ---------------#

from uuid import uuid4
from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy.orm import relationship

def generate_uuid():
    return str(uuid4())

class NotificationGroup(Base):
    __tablename__ = "notification_group"

    id = Column("id", String(75), primary_key=True, default=generate_uuid)
    client_id = Column('client_id', Integer, nullable=False)
    denotation = Column('denotation', String(250), nullable=False) 
    description = Column('description', String(500))
    customers_sites = Column('customers_sites', Text, nullable=False)
    group_mappings = relationship("NotificationGroupMapping", backref="notification_group_mapping", lazy='dynamic')

class NotificationGroupMapping(Base):
    __tablename__ = "notification_group_mapping"

    id = Column("id", String(75), primary_key=True, default=generate_uuid)
    notification_group_id = Column(String(75), ForeignKey('notification_group.id'))
    event_id = Column(String(75), nullable=False)
    recipient_id = Column(String(75), ForeignKey('recipient_group.id'))
    recipient = relationship("Recipient")
    is_used = Column(String(75), nullable=False)

class Recipient(Base):
    __tablename__ = 'recipients'

    client_id = Column('client_id', Integer, nullable=False)
    user_id = Column('user_id', Integer, nullable=False)
    email = Column('email', String(256), nullable=False)

#-------------- Query -------------#

from sqlalchemy_json_querybuilder.querybuilder.search import Search

# `filter_by` can have multiple criteria objects bundled as a list.
filter_by = [{
    "field_name": "NotificationGroup.group_mappings",
    "field_value": {
      "field_name": "NotificationGroupMapping.recipient",
      "field_value": {
        "field_name": "Recipient.email",
        "field_value": "[e-mail]",
        "operator": "equals"
      },
      "operator": "has"
    },
    "operator": "any"
}]

# `order_by` can have multiple column names. `-` indicates arranging the results in `DESC` order.
order_by = ['-NotificationGroup.client_id']

results = Search(session, "models.notification_group", (NotificationGroup,), filter_by=filter_by, order_by=order_by)

# Above code snippet is equivalent to

results = session.query(NotificationGroup).filter(
            NotificationGroup.group_mappings.any(
                NotificationGroupMapping.recipient.has(
                    Recipient.email=='[e-mail]'
                )
            )
          ).all()

Examples

Examples can be found here

TODO

and, or, match operators support.

References

• Relationship operators
• Filter operators