As of SQLAlchemy 1.4.25 there is no built-in support so I'm trying to use the solution here. This is my replication:
from datetime import date
from sqlalchemy import Column, Integer, Date, func, select, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql.elements import Over
class RangeDays(str):
def __new__(cls, x):
obj = super().__new__(cls, f"{abs(x)} day" if abs(x) == 1 else f"{abs(x)} days")
obj.x = x
return obj
def __abs__(self):
# abs(range_[0]) called in SQLCompiler._format_frame_clause
return self
def __lt__(self, other):
# range_[0] < 0 called in SQLCompiler._format_frame_clause
return self.x.__lt__(other)
_old_interpret_range = Over._interpret_range
def _interpret_range(self, range_):
lower, lower_ = (None, range_[0]) if isinstance(range_[0], RangeDays)
else (range_[0], None)
upper, upper_ = (None, range_[1]) if isinstance(range_[1], RangeDays)
else (range_[1], None)
lower, upper = _old_interpret_range(self, (lower, upper))
return lower_ or lower, upper_ or upper
Over._interpret_range = _interpret_range
Base = declarative_base()
class A(Base):
__tablename__ = "a"
__table_args__ = {"schema": "test"}
id_ = Column(Integer, primary_key=True)
category = Column(Integer)
record_date = Column(Date)
example_value = Column(Integer)
engine = create_engine(<my_connection_str>)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(A(record_date=date(2022, 5 ,1), category=1, example_value=1))
session.add(A(record_date=date(2022, 5 ,2), category=1, example_value=2))
session.add(A(record_date=date(2022, 5 ,3), category=1, example_value=3))
session.commit()
stmt = select(
func.SUM(A.example_value).over(
partition_by=A.category,
order_by=A.record_date,
range_=(RangeDays(-2), RangeDays(-1)),
)
)
session.execute(stmt).scalar()
However, when trying this with a MySQL backend I get the error:
Exception has occurred: ProgrammingError
(mysql.connector.errors.ProgrammingError)
Failed processing pyformat-parameters;
Python 'rangedays' cannot be converted to a MySQL type
[SQL: SELECT sum(test.a.example_value)
OVER (PARTITION BY test.a.category ORDER BY test.a.record_date
RANGE BETWEEN %(param_1)s PRECEDING AND %(param_2)s PRECEDING) AS anon_1
FROM test.a]
[parameters: {'param_1': '2 days', 'param_2': '1 day'}]
I've tried adapting the solution to add INTERVAL before the integers and use only DAY afterwards:
obj = super().__new__(cls, f"INTERVAL {abs(x)} DAY" if abs(x) == 1
else f"INTERVAL {abs(x)} DAY")
However, I get the same kind of error. I've tried created the strings using literal also to no avail.
Any ideas on where I could go from here?
Update:
The above should return the following SQL:
SELECT SUM(example_value) OVER (PARTITION BY category
ORDER BY record_date
RANGE BETWEEN INTERVAL 2 DAY PRECEDING AND INTERVAL 1 DAY PRECEDING)
FROM test.a;
And the following output:
+------+
| NULL |
| 1 |
| 3 |
+------+