SQLalchemy - Build ORM model from query

user_4360909 0
Viewed 2342

We have an application which uses sqlalchemy as ORM. Apart from mapping tables, we also use it to map existing database views, which works fine.

However, we cannot find an easy way to map custom queries to Models. For example, let's say we have a report like:

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
)

We would like to map this query to a class like

class Report(CustomBaseModel):

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

Our CustomBaseModel has a lot of functionality already implemented, so being able to use it here would lead to less code duplication. For instance, using the declarative base this way would make it possible to define relationships with to other tables.

Of course this example doesn't work because there is no selectable "report" in the database.

I tried setting __table__ = q.subquery() in the Report class. This way the query works but it is not possible to define columns, relationships or other properties. Perhaps there is a better way to achieve this.

flask-sqlalchemy sqlalchemy python
1 Answers

I dont quite see what kind of behavior are you want to achive, in this example the table Report exist but dosen't needs to, it's to have some data to query in a simplified way, my_query() and my_query_session() make use of a wraper that treats it as a column in the idea that you call the function name in the query and it outputs a result.

@hybrid_property can't not accept *args neither **kwargs and @hybrid_method will.

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))
user_114206300
Related