的SQLAlchemy：修改从查询对象

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。关于工会（相同的列数等）的正常注意事项都适用于这里更多（需要过滤相同的列名称）。另外，我不认为我会永远在实践中使用这个....

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]