python - SQLalchemy - 来自查询的模型

Question

关闭。这个问题需要debugging details .它目前不接受答案。












想改善这个问题吗？更新问题，使其成为 on-topic对于堆栈溢出。

3个月前关闭。




Improve this question




我们有一个应用程序使用 sqlalchemy作为 ORM。除了映射表，我们还用它来映射现有的数据库 View ，效果很好。

但是，我们找不到将自定义查询映射到模型的简单方法。例如，假设我们有一个像这样的报告:

q = session.query(
    Table1.field1, Table2.field2, func.sum(Table2.field3).label('sum')
).join(
    Table2, Table1.id == Table2.id
).filter(
    Table1.field1 == param1, Table1.field2 == param2
).group_by(
    Table1.field1, Table2.field2
) 

我们想将此查询映射到一个类

class Report(CustomBaseModel):

    field1 = Column(Integer, primary_key=True)
    field2 = Column(Integer, primary_key=True)
    sum = Column(Numeric)

我们的 CustomBaseModel已经实现了很多功能，因此能够在这里使用它会减少代码重复。例如，以这种方式使用声明性基类可以定义 relationship与其他表。

当然这个例子不起作用，因为没有可选的 "report"在数据库中。

我试过设置 __table__ = q.subquery()在 Report类(class)。通过这种方式，查询可以工作，但无法定义列、关系或其他属性。也许有更好的方法来实现这一目标。

Answer 1

我不太明白你想要实现什么样的行为，在这个例子中表 Report存在但不需要，就是有一些数据可以用简化的方式查询，my_query()和 my_query_session()使用将其视为 column 的包装器在您调用查询中的函数名称并输出结果的想法中。
@hybrid_property不能不接受*args都不是 **kwargs和 @hybrid_method将要。

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
from sqlalchemy.orm import sessionmaker
import sqlalchemy as sql

Base = declarative_base()
db_uri = 'sqlite:///test.db'
engine = sql.create_engine(db_uri)
Session = sessionmaker(bind=engine)
session = Session()

class Report(Base):
    __tablename__ = 'report'
    id = sql.Column(sql.Integer, primary_key=True)
    name = sql.Column(sql.String(30), default='Unknown')

    @hybrid_property
    def my_query(self):
        return self.name

    @hybrid_method
    def my_query_session(self, _session):
        return _session.query(Report).order_by(
            Report.id.desc()).first().name


if __name__ == '__main__':
    Report.__table__.create(engine)
    session.add(Report(name='Alex'))
    session.add(Report(name='Nata'))
    session.commit()

    query_data = session.query(Report).first()
    print(query_data.my_query)
    print(query_data.my_query_session(session))