我有几个具有相同抽象基础和相同模式的类,它们引用数据库中的类似表。我的查询非常直截了当,没有连接,简单直接的过滤条件。我在类层次结构中使用多态身份,因此我可以无缝地执行联合。
问题是有时我需要对多个表重复相同的查询并执行联合。我在 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
方法的东西,或者某种实现它的方法?
不用说,如果有更好的方法,我会洗耳恭听。
最佳答案
据我所知/根据我的经验/根据这个 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]
关于python - SQLAlchemy:修改查询对象的 from 子句,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/26429131/