Session.query_property

在 SQLAlchemy 中可以使用 session 的方法query_property方便的使用自定义的 query 类 ^[1].

以下是官方的示例代码

Session = scoped_session(sessionmaker())

class MyClass(object):
    query = Session.query_property()

# after mappers are defined
result = MyClass.query.filter(MyClass.name=='foo').all()

query_property接收一个参数cls, 指定自定义的Query class (用户从sqlalchemy.orm.Query继承的子类), 而不是默认的Query.

示例

from sqlalchemy.orm import Query

class MyClassQuery(Query):
    pass

class MyClass(object):
    query = Session.query_property(cls=MyClassQuery)

# after mappers are defined
assert isinstance(MyClass.query, MyClassQuery)

以上的方法可以完成一些基本的需求, 但是问题是MyClass.query的类定义时和Session耦合在一起了.

假如我想使用自定义的Query类, 使用query构造出查询条件后,需要应用到不同的session上 (也许几个session代表不同的数据库连接), 这就要求My Class.query在定义时需要与 Session解耦, 最后使用query.with_session方法将query绑定到session上应用查询.

自定义Query类, 与Session解耦

1	例子使用了__init_subclass__方法, 要求python3.6+. 3.6之前的版本需要进行相应的修改.

from sqlalchemy import (
    Column,
    String,
    Integer,
    and_,
    orm,
    create_engine
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.orm.base import class_mapper, exc as orm_exc


class BaseQuery(orm.Query):

    def __call__(self, with_default=False, **kwargs):
        model_cls = self._mapper_zero().class_
        if with_default:
            filters = model_cls._default_filters()
            if filters is not None:
                self = self.enable_assertions(False).filter(filters)
        return self


class UserQuery(BaseQuery):
    pass


class QueryPropertyMixin:
    query_cls = BaseQuery
    query: BaseQuery = None

    @classmethod
    def _default_filters(cls):
        pass

    @staticmethod
    def _query_property(query_cls):

        class query(object):
            def __get__(self, instance, owner):
                try:
                    mapper = class_mapper(owner)
                    if mapper:
                        # custom query class
                        return query_cls(mapper)
                except orm_exc.UnmappedClassError:
                    return None

        return query()

    def __init_subclass__(cls, **kwargs):
        cls.query = QueryPropertyMixin._query_property(cls.query_cls)


Base = declarative_base()


class User(Base, QueryPropertyMixin):
    __tablename__ = 'users'

    query_cls = UserQuery

    @classmethod
    def _default_filters(cls):
        return and_(
            cls.name != None,
            cls.fullname != None
        )

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)


engine = create_engine('sqlite:///:memory:', echo=False)

Session = scoped_session(sessionmaker(bind=engine))

session = Session()
Base.metadata.create_all(engine)

session.add(User(id=1, name='name', fullname='fullname'))
session.commit()

assert isinstance(User.query, UserQuery)

record = User.query.filter_by(id=1).with_session(session).one()
assert record.id == 1

query_str = User.query(with_default=True)
assert str(query_str).find('User.name IS NOT NULL')

首先定义自己的Query继承关系, BaseQuery继承orm.Query, 每个model如果需要有自己的特定查询类则继承BaseQuery. 在User中使用query_cls指定特定的查询类, 不定义则使用默认的BaseQuery.

核心点在于User模型继承了QueryPropertyMixin, 在QueryPropertyMixin使用了__init_subclass__方法控制子类在初始化时生成query对象.这样User.query对象就与session解耦, 在之后就使用with_session与具体session绑定进行查询了.

例子中还演示了如何在model类定义中一个default_filters, 代表某些情况中需要使用的 filters, 方便查询特定条件的对象.

SQLAlchemy session query_property ↩