@@ -1,3 +1,4 @@
|
||||
* Add support for AGE to SQLite backend
|
||||
* Add button to mark notification as read
|
||||
|
||||
Version 8.0.0 - 2026-04-20
|
||||
|
||||
@@ -218,6 +218,30 @@ def _date_trunc_interval(field, source):
|
||||
return adapt_timedelta(value)
|
||||
|
||||
|
||||
def age(timestamp1, timestamp2=None):
|
||||
def convert(value):
|
||||
for fromisoformat in [
|
||||
dt.date.fromisoformat,
|
||||
dt.datetime.fromisoformat,
|
||||
]:
|
||||
try:
|
||||
value = fromisoformat(value)
|
||||
except ValueError:
|
||||
continue
|
||||
break
|
||||
else:
|
||||
raise ValueError
|
||||
if not isinstance(value, dt.datetime):
|
||||
value = dt.datetime.combine(value, dt.time())
|
||||
return value
|
||||
timestamp1 = convert(timestamp1)
|
||||
if timestamp2 is None:
|
||||
timestamp2 = dt.datetime.combine(dt.date.today(), dt.time())
|
||||
else:
|
||||
timestamp1, timestamp2 = convert(timestamp2), timestamp1
|
||||
return adapt_timedelta(timestamp2 - timestamp1)
|
||||
|
||||
|
||||
def split_part(text, delimiter, count):
|
||||
if text is None:
|
||||
return None
|
||||
@@ -473,6 +497,8 @@ class Database(DatabaseInterface):
|
||||
factory=SQLiteConnection)
|
||||
self._conn.create_function('extract', 2, SQLiteExtract.extract)
|
||||
self._conn.create_function('date_trunc', 2, date_trunc)
|
||||
self._conn.create_function('age', 1, age)
|
||||
self._conn.create_function('age', 2, age)
|
||||
self._conn.create_function('split_part', 3, split_part)
|
||||
self._conn.create_function('to_char', 2, to_char)
|
||||
self._conn.create_function('now', 0, now)
|
||||
|
||||
@@ -5,7 +5,7 @@ import math
|
||||
from decimal import Decimal
|
||||
|
||||
from sql import Cast, Literal, Select, Table, functions
|
||||
from sql.functions import CurrentTimestamp, DateTrunc, Extract, ToChar
|
||||
from sql.functions import Age, CurrentTimestamp, DateTrunc, Extract, ToChar
|
||||
|
||||
from trytond import backend
|
||||
from trytond.model import fields
|
||||
@@ -269,6 +269,47 @@ class BackendTestCase(TestCase):
|
||||
value, = cursor.fetchone()
|
||||
self.assertEqual(value, None)
|
||||
|
||||
@with_transaction()
|
||||
def test_function_age(self):
|
||||
"Test Age function"
|
||||
cursor = Transaction().connection.cursor()
|
||||
|
||||
for t1, t2, result in [
|
||||
(dt.date(2026, 4, 13), dt.date(2026, 4, 1),
|
||||
dt.timedelta(days=12)),
|
||||
(dt.date(2026, 4, 13), dt.datetime(2026, 4, 1, 12, 0),
|
||||
dt.timedelta(days=11, hours=12)),
|
||||
(dt.datetime(2026, 4, 13, 12, 0), dt.date(2026, 4, 1),
|
||||
dt.timedelta(days=12, hours=12)),
|
||||
(dt.datetime(2026, 4, 13, 12, 0),
|
||||
dt.datetime(2026, 4, 1, 12, 0),
|
||||
dt.timedelta(days=12)),
|
||||
]:
|
||||
for t1, t2, result in [
|
||||
(t1, t2, result),
|
||||
(t2, t1, -result),
|
||||
]:
|
||||
with self.subTest(t1=t1, t2=t2):
|
||||
query = Select([Age(t1, t2).as_('age')])
|
||||
if backend.name == 'sqlite':
|
||||
sqlite_apply_types(query, ['INTERVAL'])
|
||||
cursor.execute(*query)
|
||||
value, = cursor.fetchone()
|
||||
self.assertEqual(value, result)
|
||||
|
||||
@with_transaction()
|
||||
def test_function_age_current_date(self):
|
||||
"Test Age function with current_date"
|
||||
cursor = Transaction().connection.cursor()
|
||||
|
||||
today = dt.date.today()
|
||||
query = Select([Age(today - dt.timedelta(days=10)).as_('age')])
|
||||
if backend.name == 'sqlite':
|
||||
sqlite_apply_types(query, ['INTERVAL'])
|
||||
cursor.execute(*query)
|
||||
value, = cursor.fetchone()
|
||||
self.assertEqual(value, dt.timedelta(days=10))
|
||||
|
||||
@with_transaction()
|
||||
def test_operator_range_contain(self):
|
||||
"Test Range Contain operator"
|
||||
|
||||
Reference in New Issue
Block a user