python - SQLAlchemy:修改查询对象的 from 子句

Question

我有几个具有相同抽象基础和相同模式的类，它们引用数据库中的类似表。我的查询非常直截了当，没有连接，简单直接的过滤条件。我在类层次结构中使用多态身份，因此我可以无缝地执行联合。

问题是有时我需要对多个表重复相同的查询并执行联合。我在 SQLAlchemy 中找不到该问题的解决方案，我正在尝试在我的自定义 BaseQuery 类上实现一个方法，通过克隆原始查询并更改用于自动执行所有这些操作的类/映射器from 子句。

例如，今天我必须做这样的事情:

query1 = MyModel1.query.filter_by(foo=bar)
query2 = MyModel2.query.filter_by(foo=bar)
query3 = MyModel3.query.filter_by(foo=bar)

query = query1.union(query2).union(query3)

我希望能够做类似的事情

query = MyModel1.query.filter_by(foo=bar).with_unions(MyModel2, MyModel3)

with_unions 会是这样的，其中 replace_from_clause 是我所追求的方法:

def with_unions(self, *others):
    query = self._clone()

    for other in others:
        query = query.union(replace_from_clause(query, other))

    return query

在 SQLAlchemy 的某处是否有类似 replace_from_clause 方法的东西，或者某种实现它的方法？

不用说，如果有更好的方法，我会洗耳恭听。

Answer 1

据我所知/根据我的经验/根据这个 StackOveflow 答案:https://stackoverflow.com/a/10612690/3329834你不能像这样与 ORM 联合。

我设法实现了您正在寻找的语法(或多或少)，并在返回时将所有内容加载回 orm。关于联合的正常注意事项(相同的列数等)都适用于 more 此处(需要过滤所依据的相同列名)。此外，我不认为我会在实践中使用它......

from functools import partial
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import *
from sqlalchemy import orm
from sqlalchemy import sql

engine = sqlalchemy.create_engine('sqlite://')
connection = engine.connect()


Base = declarative_base()


class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True)
    name = Column(String(767), unique=True)
    caretaker = Column(String(50))

    def __repr__(self):
        return 'Student(name={s.name}, caretaker={s.caretaker}'.format(s=self)


class Patient(Base):
    __tablename__ = "patients"
    id = Column(Integer, primary_key=True)
    name = Column(String(767), unique=True)
    caretaker = Column(String(50))

    def __repr__(self):
        return 'Patient(name={s.name}, caretaker={s.caretaker}'.format(s=self)

class StagedOperation(object):

    def __init__(self, attr):
        self.attr = attr

    def __call__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs


class StagedQuery(object):

    def __init__(self, model, session=None):
        self.session = session
        self.models = [model]
        self.columns = [e.name for e in model.__table__.columns]
        self.ops = []

    def __getattr__(self, attr):
        # __getattr__ fires only when an attribute is requested & not found
        # We will attempt to pass on any attribute call on to the resulting 
        # Query objects; do note this will only work, technically and logicaly, 
        # with method calls, not attribute access 
        if hasattr(orm.query.Query, attr):
            obj = StagedOperation(attr)
            self.ops.append(obj)

            # really getting hacky to enable "chaining"
            # Could also build this into the StagedOperation.__call__
            def _allow_chaining(desired_return, op, *args, **kwargs):
                op(*args, **kwargs)
                return desired_return

            return partial(_allow_chaining, self, obj)

    def with_unions(self, *models):
        self.models.extend(models)
        return self

    def with_session(self, session):
        self.session = session
        return self

    def query(self):
        q = None
        for model in self.models:
            id_col = sql.literal(model.__tablename__).label('tablename')
            columns = self.columns + [id_col]
            mq = orm.query.Query(columns).select_from(model)
            for op in self.ops:
                mq = getattr(mq, op.attr)(*op.args, **op.kwargs)
            q = q.union(mq) if q else mq
        return q

    def _deserialize_row(self, row):
        ref = {e.__tablename__: e for e in self.models}
        return ref[row.tablename](**{k: getattr(row, k) for k in self.columns})

    def one(self):
        return self._deserialize_row(
            self.query().with_session(self.session).one())

    def first(self):
        r = self.query().with_session(self.session).first()
        if r:
            return self._deserialize_row(r)

    def all(self):
        return [
            self._deserialize_row(e) for e in
            self.query().with_session(self.session).all()]


if __name__ == '__main__':
    engine = create_engine('sqlite://')
    Session = orm.sessionmaker()
    Session.configure(bind=engine)
    Base.metadata.bind = engine
    Base.metadata.create_all()

    session = Session()

    #
    # Insert some objects
    #

    stu = Student(id=1, name='John', caretaker='Mother')
    stu2 = Student(id=2, name='Sally', caretaker='Mother')
    stu3 = Student(id=3, name='Scott', caretaker='Father')

    pat = Patient(id=1, name='Susan', caretaker='Mother')
    pat2 = Patient(id=2, name='Sally', caretaker='Father')
    pat3 = Patient(id=3, name='Turnip', caretaker='Father')

    session.add_all([stu, stu2, stu3, pat, pat2, pat3])
    session.flush()

    # Some usage options
    print (
        StagedQuery(Student)
        .filter_by(caretaker='Mother')
        .with_unions(Patient)
        .with_session(session)
        .all())

    print (
        StagedQuery(Student, session=session)
        .filter_by(caretaker='Mother')
        .filter_by(name='Sally')
        .with_unions(Patient)
        .all())

打印...

[Student(name=John, caretaker=Mother, Patient(name=Susan, caretaker=Mother, Student(name=Sally, caretaker=Mother]
[Student(name=Sally, caretaker=Mother]