tryton-do/tryton
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add support for EXTRACT from INTERVAL to SQLite backend

Browse Source 
Closes #13305
This commit is contained in:
Cédric Krier
2024-12-14 16:22:59 +01:00
parent c31522734f
commit 81cd587477
3 changed files with 165 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trytond/CHANGELOG
+1
View File
@@ -1,3 +1,4 @@
* Add support for EXTRACT from INTERVAL to SQLite backend
* Add msg_gettext and msg_ngettext to Report context
* Support plural translation of messages
* Support plural translations in report
trytond/trytond/backend/sqlite/database.py
+104 -36
View File
@@ -1,13 +1,13 @@
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
import datetime
import datetime as dt
import logging
import math
import os
import random
import sqlite3 as sqlite
import threading
import time
import urllib.parse
import warnings
from decimal import Decimal
@@ -51,44 +51,112 @@ class SQLiteExtract(Function):
_function = 'EXTRACT'
@staticmethod
def extract(lookup_type, date):
if date is None:
def extract(lookup_type, source):
if source is None:
return None
if len(date) == 10:
year, month, day = map(int, date.split('-'))
date = datetime.date(year, month, day)
if isinstance(source, (int, float)): # interval
source = dt.timedelta(seconds=source)
else:
datepart, timepart = date.split(" ")
year, month, day = map(int, datepart.split("-"))
timepart_full = timepart.split(".")
hours, minutes, seconds = map(int, timepart_full[0].split(":"))
if len(timepart_full) == 2:
microseconds = int(timepart_full[1])
for fromisoformat in [
dt.date.fromisoformat,
dt.time.fromisoformat,
dt.datetime.fromisoformat,
]:
try:
source = fromisoformat(source)
except ValueError:
continue
break
else:
microseconds = 0
date = datetime.datetime(year, month, day, hours, minutes, seconds,
microseconds)
if lookup_type.lower() == 'century':
return date.year / 100 + (date.year % 100 and 1 or 0)
elif lookup_type.lower() == 'decade':
return date.year / 10
elif lookup_type.lower() == 'dow':
return (date.weekday() + 1) % 7
elif lookup_type.lower() == 'doy':
return date.timetuple().tm_yday
elif lookup_type.lower() == 'epoch':
return int(time.mktime(date.timetuple()))
elif lookup_type.lower() == 'microseconds':
return date.microsecond
elif lookup_type.lower() == 'millennium':
return date.year / 1000 + (date.year % 1000 and 1 or 0)
elif lookup_type.lower() == 'milliseconds':
return date.microsecond / 1000
elif lookup_type.lower() == 'quarter':
return date.month / 4 + 1
elif lookup_type.lower() == 'week':
return date.isocalendar()[1]
return getattr(date, lookup_type.lower())
raise ValueError
lookup_type = lookup_type.lower()
if lookup_type == 'century':
if isinstance(source, dt.date):
return source.year // 100 + (source.year % 100 and 1 or 0)
elif isinstance(source, dt.timedelta):
return 0
elif lookup_type == 'day':
if isinstance(source, dt.date):
return source.day
elif isinstance(source, dt.timedelta):
return source.days
elif lookup_type == 'decade':
if isinstance(source, dt.date):
return source.year // 10
elif lookup_type == 'dow':
if isinstance(source, dt.date):
return (source.weekday() + 1) % 7
elif lookup_type == 'doy':
if isinstance(source, dt.date):
return source.timetuple().tm_yday
elif lookup_type == 'epoch':
if isinstance(source, dt.datetime):
return source.timestamp()
elif isinstance(source, dt.date):
return int(dt.datetime.combine(source, dt.time()).timestamp())
elif isinstance(source, dt.timedelta):
return (
source.days * 24 * 60 * 60
+ source.seconds
+ source.microseconds / 1_000_000)
elif lookup_type == 'hour':
if isinstance(source, (dt.datetime, dt.time)):
return source.hour
elif isinstance(source, dt.timedelta):
return source.seconds // 60 // 60
elif lookup_type == 'isodow':
if isinstance(source, dt.date):
return source.isoweekday()
elif lookup_type == 'isoyear':
if isinstance(source, dt.date):
return source.isocalendar()[0]
elif lookup_type == 'microseconds':
if isinstance(source, (dt.datetime, dt.time)):
return source.second * 1_000_000 + source.microsecond
elif isinstance(source, dt.timedelta):
return (source.seconds % 60) * 1_000_000 + source.microseconds
elif lookup_type == 'millennium':
if isinstance(source, dt.date):
return source.year // 1000 + (source.year % 1000 and 1 or 0)
elif isinstance(source, dt.timedelta):
return 0
elif lookup_type == 'milliseconds':
if isinstance(source, (dt.datetime, dt.time)):
return source.second * 1_000 + source.microsecond / 1_000
elif isinstance(source, dt.timedelta):
return (
(source.seconds % 60) * 1_000
+ source.microseconds / 1_000)
elif lookup_type == 'minute':
if isinstance(source, (dt.datetime, dt.time)):
return source.minute
elif isinstance(source, dt.timedelta):
return source.seconds // 60
elif lookup_type == 'month':
if isinstance(source, dt.date):
return source.month
elif isinstance(source, dt.timedelta):
return 0
elif lookup_type == 'quarter':
if isinstance(source, dt.date):
return source.month // 4 + 1
elif isinstance(source, dt.timedelta):
return 1
elif lookup_type == 'second':
if isinstance(source, (dt.datetime, dt.time)):
return source.second + source.microsecond / 1_000_000
elif isinstance(source, dt.timedelta):
return (source.seconds % 60) + source.microseconds / 1_000_000
elif lookup_type == 'week':
if isinstance(source, dt.date):
return source.isocalendar()[1]
elif lookup_type == 'year':
if isinstance(source, dt.date):
return source.year
elif isinstance(source, dt.timedelta):
return 0
raise ValueError
def date_trunc(_type, date):
trytond/trytond/tests/test_backend.py
+60 -1
View File
@@ -2,9 +2,10 @@
# this repository contains the full copyright notices and license terms.
import datetime as dt
import math
from decimal import Decimal
from sql import Cast, Literal, Select, Table, functions
from sql.functions import CurrentTimestamp, DateTrunc, ToChar
from sql.functions import CurrentTimestamp, DateTrunc, Extract, ToChar
from trytond.model import fields
from trytond.sql.functions import NumRange
@@ -280,3 +281,61 @@ class BackendTestCase(TestCase):
count = database.estimated_count(connection, query)
self.assertEqual(count, 1)
@with_transaction()
def test_function_extract(self):
"Test Extract function"
cursor = Transaction().connection.cursor()
for lookup, source, result in [
('CENTURY', dt.datetime(2000, 12, 16, 12, 21, 13), 20),
('CENTURY', dt.datetime(2001, 2, 16, 20, 38, 40), 21),
('CENTURY', dt.date(1, 1, 1), 1),
('CENTURY', dt.timedelta(days=2001 * 365), 0),
('DAY', dt.datetime(2001, 2, 16, 20, 38, 40), 16),
('DAY', dt.timedelta(40, 1), 40),
('DECADE', dt.datetime(2001, 2, 16, 20, 38, 40), 200),
('DOW', dt.datetime(2001, 2, 16, 20, 38, 40), 5),
('DOY', dt.datetime(2001, 2, 16, 20, 38, 40), 47),
('EPOCH', dt.datetime(2001, 2, 16, 20, 38, 40, 120000),
982355920.120000),
('EPOCH', dt.date(2001, 2, 16), 982281600.000000),
('EPOCH', dt.timedelta(days=5, hours=3), 442800.000000),
('HOUR', dt.datetime(2001, 2, 16, 20, 38, 40), 20),
('HOUR', dt.timedelta(hours=20, minutes=38, seconds=40), 20),
('ISODOW', dt.datetime(2001, 2, 18, 20, 38, 40), 7),
('ISOYEAR', dt.date(2006, 1, 1), 2005),
('ISOYEAR', dt.date(2006, 1, 2), 2006),
('MICROSECONDS', dt.time(17, 12, 28, 500000), 28500000),
('MICROSECONDS', dt.timedelta(hours=1, seconds=28.5),
28500000),
('MILLENNIUM', dt.datetime(2001, 2, 16, 20, 38, 40), 3),
('MILLENNIUM', dt.timedelta(2001 * 365), 0),
('MILLISECONDS', dt.time(17, 12, 28, 500000), 28500.000),
('MILLISECONDS', dt.timedelta(seconds=28, microseconds=500000),
28500.000),
('MINUTE', dt.datetime(2001, 2, 16, 20, 38, 40), 38),
('MINUTE', dt.timedelta(seconds=60), 1),
('MINUTE', dt.timedelta(seconds=61), 1),
('MINUTE', dt.timedelta(seconds=120), 2),
('MONTH', dt.datetime(2001, 2, 16, 20, 38, 40), 2),
('MONTH', dt.timedelta(60), 0),
('QUARTER', dt.datetime(2001, 2, 16, 20, 38, 40), 1),
('QUARTER', dt.timedelta(1), 1),
('QUARTER', dt.timedelta(365), 1),
('SECOND', dt.datetime(2001, 2, 16, 20, 38, 40), 40.000000),
('SECOND', dt.time(17, 12, 28, 500000), 28.500000),
('SECOND', dt.timedelta(
hours=17, minutes=12, seconds=28, microseconds=500000),
28.500000),
('WEEK', dt.datetime(2001, 2, 16, 20, 38, 40), 7),
('YEAR', dt.datetime(2001, 2, 16, 20, 38, 40), 2001),
('YEAR', dt.timedelta(days=366), 0),
]:
for lookup in [lookup.lower(), lookup.upper()]:
with self.subTest(lookup=lookup, source=source):
cursor.execute(*Select([Extract(lookup, source)]))
value, = cursor.fetchone()
# in PostgreSQL some value are numeric
if isinstance(value, Decimal):
value = float(value)
self.assertEqual(value, result)