的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]