fix: add DateTimeAlchemyFilter for datetime columns
This commit is contained in:
parent
f8cb97ce63
commit
20a586e950
2 changed files with 397 additions and 7 deletions
|
|
@ -2,7 +2,7 @@
|
|||
################################################################################
|
||||
#
|
||||
# wuttaweb -- Web App for Wutta Framework
|
||||
# Copyright © 2024-2025 Lance Edgar
|
||||
# Copyright © 2024-2026 Lance Edgar
|
||||
#
|
||||
# This file is part of Wutta Framework.
|
||||
#
|
||||
|
|
@ -682,9 +682,12 @@ class BooleanAlchemyFilter(AlchemyFilter):
|
|||
class DateAlchemyFilter(AlchemyFilter):
|
||||
"""
|
||||
SQLAlchemy filter option for a
|
||||
:class:`sqlalchemy:sqlalchemy.types.Date` column.
|
||||
:class:`~sqlalchemy:sqlalchemy.types.Date` column.
|
||||
|
||||
Subclass of :class:`AlchemyFilter`.
|
||||
|
||||
This filter class has custom logic to coerce the input value, but
|
||||
does not have custom filter logic beyond that.
|
||||
"""
|
||||
|
||||
data_type = "date"
|
||||
|
|
@ -706,10 +709,23 @@ class DateAlchemyFilter(AlchemyFilter):
|
|||
"less_than": "before",
|
||||
"less_equal": "on or before",
|
||||
# 'between': "between",
|
||||
"is_any": "is any",
|
||||
}
|
||||
|
||||
def coerce_value(self, value): # pylint: disable=empty-docstring
|
||||
""" """
|
||||
def coerce_value(self, value):
|
||||
"""
|
||||
Convert the given value to a proper
|
||||
:class:`python:datetime.date` object, if applicable.
|
||||
|
||||
If the input value is already a date object, it is returned
|
||||
as-is.
|
||||
|
||||
Otherwise it is assumed to be a string in ``%Y-%m-%d`` format,
|
||||
and will be converted to a date object.
|
||||
|
||||
If the conversion fails, or no value is provided, ``None`` is
|
||||
returned.
|
||||
"""
|
||||
if value:
|
||||
if isinstance(value, datetime.date):
|
||||
return value
|
||||
|
|
@ -724,6 +740,145 @@ class DateAlchemyFilter(AlchemyFilter):
|
|||
return None
|
||||
|
||||
|
||||
class DateTimeAlchemyFilter(DateAlchemyFilter):
|
||||
"""
|
||||
SQLAlchemy filter option for a
|
||||
:class:`~sqlalchemy:sqlalchemy.types.DateTime` column.
|
||||
|
||||
Subclass of :class:`DateAlchemyFilter`.
|
||||
|
||||
This filter class has custom logic to coerce the input value,
|
||||
inherited from parent class. It also has custom filter logic
|
||||
for most verbs (not/equal, greater/less than etc.).
|
||||
|
||||
Please note that this class assumes the underlying data uses
|
||||
"naive UTC" values. It therefore will convert to/from local time
|
||||
zone accordingly, to ensure user gets the data they expect. For
|
||||
more info see :doc:`wuttjamaican:narr/datetime`.
|
||||
"""
|
||||
|
||||
def get_start_datetime(self, value, as_utc=True):
|
||||
"""
|
||||
Calculate the "start" timestamp for the given date value.
|
||||
|
||||
The return value will be the "first possible moment" of the
|
||||
given date.
|
||||
|
||||
:param value: :class:`python:datetime.date` instance
|
||||
|
||||
:param as_utc: Indicates the return value should be naive/UTC;
|
||||
set this to ``False`` to get the aware/local value.
|
||||
|
||||
:returns: :class:`python:datetime.datetime` instance
|
||||
"""
|
||||
start = datetime.datetime.combine(value, datetime.time(0))
|
||||
start = self.app.localtime(start, from_utc=False)
|
||||
if as_utc:
|
||||
start = self.app.make_utc(start)
|
||||
return start
|
||||
|
||||
def get_end_datetime(self, value, as_utc=True):
|
||||
"""
|
||||
Calculate the "end" timestamp for the given date value.
|
||||
|
||||
Due to the nature of queries involving this "end" boundary,
|
||||
the return value will be the "first possible moment" of the
|
||||
day *after* the given date.
|
||||
|
||||
:param value: :class:`python:datetime.date` instance
|
||||
|
||||
:param as_utc: Indicates the return value should be naive/UTC;
|
||||
set this to ``False`` to get the aware/local value.
|
||||
|
||||
:returns: :class:`python:datetime.datetime` instance
|
||||
"""
|
||||
end = datetime.datetime.combine(
|
||||
value + datetime.timedelta(days=1), datetime.time(0)
|
||||
)
|
||||
end = self.app.localtime(end, from_utc=False)
|
||||
if as_utc:
|
||||
end = self.app.make_utc(end)
|
||||
return end
|
||||
|
||||
def filter_equal(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall on the given
|
||||
date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
start = self.get_start_datetime(value)
|
||||
end = self.get_end_datetime(value)
|
||||
return query.filter(self.model_property >= start).filter(
|
||||
self.model_property < end
|
||||
)
|
||||
|
||||
def filter_not_equal(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall outside the
|
||||
given date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
start = self.get_start_datetime(value)
|
||||
end = self.get_end_datetime(value)
|
||||
return query.filter(
|
||||
sa.or_(self.model_property < start, self.model_property >= end)
|
||||
)
|
||||
|
||||
def filter_greater_than(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall after the
|
||||
given date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
end = self.get_end_datetime(value)
|
||||
return query.filter(self.model_property >= end)
|
||||
|
||||
def filter_greater_equal(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall on or after
|
||||
the given date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
start = self.get_start_datetime(value)
|
||||
return query.filter(self.model_property >= start)
|
||||
|
||||
def filter_less_than(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall before the
|
||||
given date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
start = self.get_start_datetime(value)
|
||||
return query.filter(self.model_property < start)
|
||||
|
||||
def filter_less_equal(self, query, value):
|
||||
"""
|
||||
Find all records with datetime values which fall on or before
|
||||
the given date.
|
||||
"""
|
||||
value = self.coerce_value(value)
|
||||
if value is None:
|
||||
return query
|
||||
|
||||
end = self.get_end_datetime(value)
|
||||
return query.filter(self.model_property < end)
|
||||
|
||||
|
||||
default_sqlalchemy_filters = {
|
||||
None: AlchemyFilter,
|
||||
sa.String: StringAlchemyFilter,
|
||||
|
|
@ -732,4 +887,5 @@ default_sqlalchemy_filters = {
|
|||
sa.Integer: IntegerAlchemyFilter,
|
||||
sa.Boolean: BooleanAlchemyFilter,
|
||||
sa.Date: DateAlchemyFilter,
|
||||
sa.DateTime: DateTimeAlchemyFilter,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from unittest.mock import patch
|
|||
import sqlalchemy as sa
|
||||
|
||||
from wuttjamaican.db.model import Base
|
||||
from wuttjamaican.util import get_timezone_by_name
|
||||
|
||||
from wuttaweb.grids import filters as mod
|
||||
from wuttaweb.testing import WebTestCase
|
||||
|
|
@ -569,7 +570,8 @@ class TestBooleanAlchemyFilter(WebTestCase):
|
|||
class TheLocalThing(Base):
|
||||
__tablename__ = "the_local_thing"
|
||||
id = sa.Column(sa.Integer(), primary_key=True, autoincrement=False)
|
||||
date = sa.Column(sa.DateTime(timezone=True), nullable=True)
|
||||
date = sa.Column(sa.Date(), nullable=True)
|
||||
timestamp = sa.Column(sa.DateTime(), nullable=True)
|
||||
|
||||
|
||||
class TestDateAlchemyFilter(WebTestCase):
|
||||
|
|
@ -604,7 +606,7 @@ class TestDateAlchemyFilter(WebTestCase):
|
|||
# null value
|
||||
self.assertIsNone(filtr.coerce_value(None))
|
||||
|
||||
# value as datetime
|
||||
# value as date
|
||||
value = datetime.date(2024, 1, 1)
|
||||
result = filtr.coerce_value(value)
|
||||
self.assertIs(value, result)
|
||||
|
|
@ -717,7 +719,7 @@ class TestDateAlchemyFilter(WebTestCase):
|
|||
self.sample_query, datetime.date(2024, 3, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_less_equal(self.sample_query, "2024-04-01")
|
||||
|
|
@ -725,6 +727,238 @@ class TestDateAlchemyFilter(WebTestCase):
|
|||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
|
||||
class TestDateTimeAlchemyFilter(WebTestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.setup_web()
|
||||
self.tzlocal = get_timezone_by_name("America/Los_Angeles")
|
||||
|
||||
self.sample_data = [
|
||||
{
|
||||
"id": 1,
|
||||
"timestamp": self.app.make_utc(
|
||||
datetime.datetime(2024, 1, 1, 1, 20, tzinfo=self.tzlocal)
|
||||
),
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
"timestamp": self.app.make_utc(
|
||||
datetime.datetime(2024, 1, 1, 23, 40, tzinfo=self.tzlocal)
|
||||
),
|
||||
},
|
||||
{
|
||||
"id": 3,
|
||||
"timestamp": self.app.make_utc(
|
||||
datetime.datetime(2024, 3, 1, 22, 10, tzinfo=self.tzlocal)
|
||||
),
|
||||
},
|
||||
{
|
||||
"id": 4,
|
||||
"timestamp": self.app.make_utc(
|
||||
datetime.datetime(2024, 3, 1, 2, 0, tzinfo=self.tzlocal)
|
||||
),
|
||||
},
|
||||
{"id": 5, "timestamp": None},
|
||||
{"id": 6, "timestamp": None},
|
||||
]
|
||||
|
||||
for thing in self.sample_data:
|
||||
thing = TheLocalThing(**thing)
|
||||
self.session.add(thing)
|
||||
self.session.commit()
|
||||
|
||||
self.sample_query = self.session.query(TheLocalThing)
|
||||
|
||||
def make_filter(self, model_property, **kwargs):
|
||||
factory = kwargs.pop("factory", mod.DateTimeAlchemyFilter)
|
||||
kwargs["model_property"] = model_property
|
||||
return factory(self.request, model_property.key, **kwargs)
|
||||
|
||||
def test_coerce_value(self):
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
|
||||
# null value
|
||||
self.assertIsNone(filtr.coerce_value(None))
|
||||
|
||||
# value as date
|
||||
value = datetime.date(2024, 1, 1)
|
||||
result = filtr.coerce_value(value)
|
||||
self.assertIs(value, result)
|
||||
|
||||
# value as string
|
||||
result = filtr.coerce_value("2024-04-01")
|
||||
self.assertIsInstance(result, datetime.date)
|
||||
self.assertEqual(result, datetime.date(2024, 4, 1))
|
||||
|
||||
# invalid
|
||||
result = filtr.coerce_value("thisinputisbad")
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_equal(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_equal(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date
|
||||
filtered_query = filtr.filter_equal(
|
||||
self.sample_query, datetime.date(2024, 3, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_equal(self.sample_query, "2024-02-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 0)
|
||||
filtered_query = filtr.filter_equal(self.sample_query, "2024-01-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
def test_not_equal(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_not_equal(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date
|
||||
filtered_query = filtr.filter_not_equal(
|
||||
self.sample_query, datetime.date(2024, 2, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_not_equal(self.sample_query, "2024-02-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
filtered_query = filtr.filter_not_equal(self.sample_query, "2024-01-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
def test_greater_than(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_greater_than(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date
|
||||
filtered_query = filtr.filter_greater_than(
|
||||
self.sample_query, datetime.date(2024, 2, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_greater_than(self.sample_query, "2024-02-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
def test_greater_equal(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_greater_equal(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date (clear of boundary)
|
||||
filtered_query = filtr.filter_greater_equal(
|
||||
self.sample_query, datetime.date(2024, 2, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as date (at boundary)
|
||||
filtered_query = filtr.filter_greater_equal(
|
||||
self.sample_query, datetime.date(2024, 3, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_greater_equal(self.sample_query, "2024-01-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
def test_less_than(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_less_than(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date
|
||||
filtered_query = filtr.filter_less_than(
|
||||
self.sample_query, datetime.date(2024, 2, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_less_than(self.sample_query, "2024-04-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
def test_less_equal(self):
|
||||
model = self.app.model
|
||||
with patch.object(self.app, "get_timezone", return_value=self.tzlocal):
|
||||
|
||||
filtr = self.make_filter(TheLocalThing.timestamp)
|
||||
self.assertEqual(self.sample_query.count(), 6)
|
||||
|
||||
# null value ignored
|
||||
filtered_query = filtr.filter_less_equal(self.sample_query, None)
|
||||
self.assertIs(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 6)
|
||||
|
||||
# value as date (clear of boundary)
|
||||
filtered_query = filtr.filter_less_equal(
|
||||
self.sample_query, datetime.date(2024, 2, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 2)
|
||||
|
||||
# value as date (at boundary)
|
||||
filtered_query = filtr.filter_less_equal(
|
||||
self.sample_query, datetime.date(2024, 3, 1)
|
||||
)
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
# value as string
|
||||
filtered_query = filtr.filter_less_equal(self.sample_query, "2024-04-01")
|
||||
self.assertIsNot(filtered_query, self.sample_query)
|
||||
self.assertEqual(filtered_query.count(), 4)
|
||||
|
||||
|
||||
class TestVerbNotSupported(TestCase):
|
||||
|
||||
def test_basic(self):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue