I've got a simple problem I can't figure out.
I've got a simple data model with three integer columns (which can be null). I want to create an hybrid attribute to get the sum of three columns. My wish is to finally get the maximal total or to filter according to this total.
I've made a minimal example.
from contextlib import contextmanager
from sqlalchemy import create_engine, func, select
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
engine = create_engine("sqlite://")
"""The database engine."""
DbSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)
"""The application session class."""
class Base(DeclarativeBase):
"""The SQLalchemy declarative base."""
class Data(Base):
__tablename__ = "data"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
x: Mapped[int | None]
y: Mapped[int | None]
z: Mapped[int | None]
@hybrid_property
def total(self):
return self.x + self.y + self.z
@total.expression
@classmethod
def total(cls):
return select(cls.x + cls.y + cls.z).label("total")
@contextmanager
def get_db() -> DbSession:
"""Return a database session."""
with DbSession() as db_is:
with db_is.begin():
yield db_is
# inner context calls db.commit(), if there were no exceptions
# outer context calls db.close()
Base.metadata.create_all(engine)
with get_db() as db:
# Add data
db.add(Data(x=1, y=1, z=None))
db.add(Data(x=None, y=2, z=2))
db.add(Data(x=3, y=3, z=3))
db.flush()
# Try to get maximal total
print(db.scalar(select(func.max(Data.total))))
# Try to get entries with total less than 5
print(db.scalars(select(Data).filter(Data.total < 5)).all())
My problems are:
The above code fails. The first statement db.scalar(select(func.max(Data.total)))
fails with the following message.
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) misuse of aggregate: max()
[SQL: SELECT max((SELECT data.x + data.y + data.z AS anon_1
FROM data)) AS max_1]
The second statement db.scalars(select(Data).filter(Data.total < 5)).all()
returns []
. Running db.scalars(select(Data.total)).all()
returns [None]
.
Can someone help me?
Ok, I figured out the problem.
The problem comes from my total expression. Instead of writing
@total.expression
@classmethod
def total(cls):
return select(cls.x + cls.y + cls.z).label("total")
I need to write
@total.expression
@classmethod
def total(cls):
return cls.x + cls.y + cls.z
The select and label are NOT necessary. Yet, this is a mystery for me as I use such pattern for other cases I need to agregate. Maybe this is linked to a subrequest matter.
In addition, the NULL problem is resolved using the coalesce
function. One more fix for Pythonic expression gives me:
class Data(Base):
__tablename__ = "data"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
x: Mapped[int | None]
y: Mapped[int | None]
z: Mapped[int | None]
@hybrid_property
def total(self):
return (self.x or 0) + (self.y or 0) + (self.z or 0)
@total.expression
@classmethod
def total(cls):
return func.coalesce(cls.x, 0) + func.coalesce(cls.y, 0) + func.coalesce(cls.z, 0)
That's the answer :)