sqlalchemy django-like smart query simple example

smartquery.py

# -*- coding: UTF-8 -*-
from __future__ import print_function
import os

import datetime
import sqlalchemy as sa
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Query, scoped_session, sessionmaker

#################### setup ######################
Base = declarative_base()


# we also use ReprMixin which is optional
class User(Base):
    __tablename__ = 'user'
    __repr_attrs__ = ['name']
    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)

    # to smart query relationship, it should be explicitly set,
    # not to be a backref
    posts = sa.orm.relationship('Post')
    comments = sa.orm.relationship('Comment')


class Post(Base):
    __tablename__ = 'post'
    id = sa.Column(sa.Integer, primary_key=True)
    body = sa.Column(sa.String)
    user_id = sa.Column(sa.Integer, sa.ForeignKey('user.id'))
    archived = sa.Column(sa.Boolean, default=False)

    # to smart query relationship, it should be explicitly set,
    # not to be a backref
    user = sa.orm.relationship('User')
    comments = sa.orm.relationship('Comment')


class Comment(Base):
    __tablename__ = 'comment'
    __repr_attrs__ = ['body']
    id = sa.Column(sa.Integer, primary_key=True)
    body = sa.Column(sa.String)
    user_id = sa.Column(sa.Integer, sa.ForeignKey('user.id'))
    post_id = sa.Column(sa.Integer, sa.ForeignKey('post.id'))
    rating = sa.Column(sa.Integer)
    created_at = sa.Column(sa.DateTime)

    # to smart query relationship, it should be explicitly set,
    # not to be a backref
    user = sa.orm.relationship('User')
    post = sa.orm.relationship('Post')


#################### setup ORM ######################

db_file = os.path.join(os.path.dirname(__file__), 'test.sqlite')
engine = create_engine('sqlite:///{}'.format(db_file), echo=True)

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

session = scoped_session(sessionmaker(bind=engine))

#################### setup some data ######################
u1 = User(name='Vasya')
session.add(u1)
session.commit()

u2 = User(name='Alex u2')
session.add(u2)
session.commit()

u3 = User(name='Bishop u3')
session.add(u3)
session.commit()

session.commit()

p11 = Post(
    id=11,
    body='1234567890123',
    archived=True,
    user=u1
)
session.add(p11)
session.commit()

p12 = Post(
    id=12,
    body='1234567890',
    user=u1
)
session.add(p12)
session.commit()

p21 = Post(
    id=21,
    body='p21',
    user=u2
)
session.add(p21)
session.commit()

p22 = Post(
    id=22,
    body='p22',
    user=u2
)
session.add(p22)
session.commit()

cm11 = Comment(
    id=11,
    body='cm11',
    user=u1,
    post=p11,
    rating=1,
    created_at=datetime.datetime(2014, 1, 1)
)
session.add(cm11)
session.commit()

cm12 = Comment(
    id=12,
    body='cm12',
    user=u2,
    post=p12,
    rating=2,
    created_at=datetime.datetime(2015, 10, 20)
)
session.add(cm12)
session.commit()

cm21 = Comment(
    id=21,
    body='cm21',
    user=u1,
    post=p21,
    rating=1,
    created_at=datetime.datetime(2015, 11, 21)
)
session.add(cm21)
session.commit()

cm22 = Comment(
    id=22,
    body='cm22',
    user=u3,
    post=p22,
    rating=3,
    created_at=datetime.datetime(2016, 11, 20)
)
session.add(cm22)
session.commit()

cm_empty = Comment(
    id=29,
    # no body
    # no user
    # no post
    # no rating
)
session.add(cm_empty)
session.commit()

#################### Demo ######################

"""
как работает
User.where(posts___comments___user___name__like='Vasya').all()
"""

from sqlalchemy.orm import aliased
from sqlalchemy.sql import operators

# Имеем на входе {'posts___comments___user___name__like': 'Vasya'}. Достанем:
relations = ['posts', 'comments', 'user'] # 1. отношения, они были разделены ___
attr_name = 'name' # 2. аттрибут, он был после последнего ___
op_name = 'like' # 3. оператор, он был после __
# получаем оператор Алхимии на основе op_name.
# в реале имеется фиксированное соответствие OPERATORS = {'like': operators.like_op},
# и из него оператор достаётся как OPERATORS[op_name]
operator = operators.like_op

value = 'Vasya'


cls = User # в жизни это статический метод и текущий класс хранится в cls
query = session.query(cls) # делаем начальный запрос

# джойним все связи в цикле
last_alias = cls
for relation in relations:
    relation = getattr(last_alias, relation)
    next_alias = aliased(relation.property.argument())
    query = query.outerjoin(next_alias)
    last_alias = next_alias

# теперь отфильтруем последнее отношение ('user')
attr = getattr(last_alias, attr_name) # получаем реальный аттрибут User.name
query = query.filter(operator(attr, value)) # применим оператор, передав ему аттрибут User.name и Васю
print(query.all())