Allow to search and order on Dict key value
issue7905 review62391002
This commit is contained in:
Cédric Krier
@@ -1,3 +1,6 @@
|
||||
* Allow to search on key's value of Dict
|
||||
* Allow to order Dict by key's value
|
||||
* Do not store keys with null value in Dict
|
||||
* Add cache timeout for web
|
||||
* Allow the bus requests to be redirected to another host
|
||||
* Add support for CORS
|
||||
|
||||
@@ -927,6 +927,10 @@ A dictionary field with predefined keys.
|
||||
The name of the :class:`DictSchemaMixin` model that stores the definition
|
||||
of keys.
|
||||
|
||||
.. attribute:: Dict.search_unaccented
|
||||
|
||||
Same as :attr:`Char.search_unaccented` but when searching on key's value.
|
||||
|
||||
Instance methods:
|
||||
|
||||
.. method:: Dict.translated([name[, type_]])
|
||||
|
||||
@@ -437,7 +437,8 @@ Class methods:
|
||||
|
||||
In case the field used is a :class:`fields.Many2One`, it is also possible
|
||||
to use the dotted notation to sort on a specific field from the target
|
||||
record.
|
||||
record. Or for a :class:`fields.Dict` field, the dotted notation is used to
|
||||
sort on the key's value.
|
||||
|
||||
If `count` is set to `True`, then the result is the number of records.
|
||||
|
||||
|
||||
+14
-6
@@ -22,15 +22,15 @@ by this pattern::
|
||||
domain = [(<field name>, <operator>, <operand>)]
|
||||
|
||||
``<field name>``
|
||||
Is the name of a :mod:`trytond.model.fields` or a
|
||||
Is the name of a :mod:`~trytond.model.fields` or a
|
||||
:ref:`pyson <topics-pyson>` statement, that evaluates to a
|
||||
string.
|
||||
|
||||
A field of type :class:`trytond.model.fields.Many2One` or
|
||||
:class:`trytond.model.fields.Many2Many` or
|
||||
:class:`trytond.model.fields.One2Many` or
|
||||
:class:`trytond.model.fields.One2One` or
|
||||
:class:`trytond.model.fields.Reference` can be dereferenced to related
|
||||
A field of type :class:`~trytond.model.fields.Many2One` or
|
||||
:class:`~trytond.model.fields.Many2Many` or
|
||||
:class:`~trytond.model.fields.One2Many` or
|
||||
:class:`~trytond.model.fields.One2One` or
|
||||
:class:`~trytond.model.fields.Reference` can be dereferenced to related
|
||||
models. This is illustrated by the following example::
|
||||
|
||||
domain = [('country.name', '=', 'Japan')]
|
||||
@@ -44,6 +44,14 @@ by this pattern::
|
||||
domain = [('origin.party.name', '=', 'John Doe', 'sale.sale')]
|
||||
..
|
||||
|
||||
A field of type :class:`~trytond.model.fields.Dict` can be searched by key
|
||||
also by using one *dot*. For example::
|
||||
|
||||
domain = [('attributes.color', '=', 'yellow')]
|
||||
|
||||
.. warning::
|
||||
Order comparison of `date` and `datetime` types is not supported.
|
||||
|
||||
``operator``
|
||||
Is an operator out of `Domain Operators`_ or a
|
||||
:ref:`pyson <topics-pyson>` statement, that evaluates to
|
||||
|
||||
@@ -230,3 +230,23 @@ class DatabaseInterface(object):
|
||||
def sql_format(self, type_, value):
|
||||
'Return value correctly casted into type_'
|
||||
pass
|
||||
|
||||
def json_get(self, column, key=None):
|
||||
"Return the JSON value of the JSON key"
|
||||
raise NotImplementedError
|
||||
|
||||
def json_key_exists(self, column, key):
|
||||
"Return expression for key exists in JSON column"
|
||||
raise NotImplementedError
|
||||
|
||||
def json_any_keys_exist(self, column, keys):
|
||||
"Return expression for any keys exist in JSON column"
|
||||
raise NotImplementedError
|
||||
|
||||
def json_all_keys_exist(self, column, keys):
|
||||
"Rteurn expression for all keys exist in JSON column"
|
||||
raise NotImplementedError
|
||||
|
||||
def json_contains(self, column, json):
|
||||
"Return expression for column contains JSON"
|
||||
raise NotImplementedError
|
||||
|
||||
@@ -29,8 +29,9 @@ from psycopg2 import IntegrityError as DatabaseIntegrityError
|
||||
from psycopg2 import OperationalError as DatabaseOperationalError
|
||||
from psycopg2.extras import register_default_json, register_default_jsonb
|
||||
|
||||
from sql import Flavor
|
||||
from sql import Flavor, Cast
|
||||
from sql.functions import Function
|
||||
from sql.operators import BinaryOperator
|
||||
|
||||
from trytond.backend.database import DatabaseInterface, SQLType
|
||||
from trytond.config import config, parse_uri
|
||||
@@ -78,6 +79,31 @@ class TryAdvisoryLock(Function):
|
||||
_function = 'pg_try_advisory_xact_lock'
|
||||
|
||||
|
||||
class JSONBExtractPath(Function):
|
||||
__slots__ = ()
|
||||
_function = 'jsonb_extract_path'
|
||||
|
||||
|
||||
class JSONKeyExists(BinaryOperator):
|
||||
__slots__ = ()
|
||||
_operator = '?'
|
||||
|
||||
|
||||
class JSONAnyKeyExist(BinaryOperator):
|
||||
__slots__ = ()
|
||||
_operator = '?|'
|
||||
|
||||
|
||||
class JSONAllKeyExist(BinaryOperator):
|
||||
__slots__ = ()
|
||||
_operator = '?&'
|
||||
|
||||
|
||||
class JSONContains(BinaryOperator):
|
||||
__slots__ = ()
|
||||
_operator = '@>'
|
||||
|
||||
|
||||
class Database(DatabaseInterface):
|
||||
|
||||
_lock = RLock()
|
||||
@@ -476,6 +502,24 @@ class Database(DatabaseInterface):
|
||||
def has_channel(self):
|
||||
return True
|
||||
|
||||
def json_get(self, column, key=None):
|
||||
column = Cast(column, 'jsonb')
|
||||
if key:
|
||||
column = JSONBExtractPath(column, key)
|
||||
return column
|
||||
|
||||
def json_key_exists(self, column, key):
|
||||
return JSONKeyExists(Cast(column, 'jsonb'), key)
|
||||
|
||||
def json_any_keys_exist(self, column, keys):
|
||||
return JSONAnyKeyExist(Cast(column, 'jsonb'), keys)
|
||||
|
||||
def json_all_keys_exist(self, column, keys):
|
||||
return JSONAllKeyExist(Cast(column, 'jsonb'), keys)
|
||||
|
||||
def json_contains(self, column, json):
|
||||
return JSONContains(Cast(column, 'jsonb'), Cast(json, 'jsonb'))
|
||||
|
||||
|
||||
register_type(UNICODE)
|
||||
if PYDATE:
|
||||
|
||||
@@ -18,7 +18,8 @@ except ImportError:
|
||||
import sqlite3 as sqlite
|
||||
from sqlite3 import IntegrityError as DatabaseIntegrityError
|
||||
from sqlite3 import OperationalError as DatabaseOperationalError
|
||||
from sql import Flavor, Table, Query, Expression, Literal
|
||||
from sql import Flavor, Table, Query, Expression, Literal, Null
|
||||
from sql.conditionals import NullIf
|
||||
from sql.functions import (Function, Extract, Position, Substring,
|
||||
Overlay, CharLength, CurrentTimestamp, Trim)
|
||||
|
||||
@@ -215,6 +216,16 @@ MAPPING = {
|
||||
}
|
||||
|
||||
|
||||
class JSONExtract(Function):
|
||||
__slots__ = ()
|
||||
_function = 'JSON_EXTRACT'
|
||||
|
||||
|
||||
class JSONQuote(Function):
|
||||
__slots__ = ()
|
||||
_function = 'JSON_QUOTE'
|
||||
|
||||
|
||||
class SQLiteCursor(sqlite.Cursor):
|
||||
|
||||
def __enter__(self):
|
||||
@@ -442,6 +453,11 @@ class Database(DatabaseInterface):
|
||||
value = int(value)
|
||||
return value
|
||||
|
||||
def json_get(self, column, key):
|
||||
if key:
|
||||
column = JSONExtract(column, '$.%s' % key)
|
||||
return NullIf(JSONQuote(column), JSONQuote(Null))
|
||||
|
||||
sqlite.register_converter('NUMERIC', lambda val: Decimal(val.decode('utf-8')))
|
||||
sqlite.register_adapter(Decimal, lambda val: str(val).encode('utf-8'))
|
||||
|
||||
|
||||
@@ -1,12 +1,21 @@
|
||||
# This file is part of Tryton. The COPYRIGHT file at the toplevel of this
|
||||
# repository contains the full copyright notices and license terms.
|
||||
from functools import partial
|
||||
import json
|
||||
|
||||
from .field import Field
|
||||
from sql import operators, Literal, Select, CombiningQuery, Cast, Null
|
||||
|
||||
from trytond import backend
|
||||
from trytond.transaction import Transaction
|
||||
from .field import Field, SQL_OPERATORS
|
||||
from ...protocols.jsonrpc import JSONDecoder, JSONEncoder
|
||||
from ...pool import Pool
|
||||
from ...tools import grouped_slice
|
||||
|
||||
# Use canonical form
|
||||
dumps = partial(
|
||||
json.dumps, cls=JSONEncoder, separators=(',', ':'), sort_keys=True)
|
||||
|
||||
|
||||
class Dict(Field):
|
||||
'Define dict field.'
|
||||
@@ -21,6 +30,7 @@ class Dict(Field):
|
||||
states, select, on_change, on_change_with, depends, context,
|
||||
loading)
|
||||
self.schema_model = schema_model
|
||||
self.search_unaccented = True
|
||||
|
||||
def get(self, ids, model, name, values=None):
|
||||
dicts = dict((id, None) for id in ids)
|
||||
@@ -37,9 +47,8 @@ class Dict(Field):
|
||||
if value is None:
|
||||
return None
|
||||
assert isinstance(value, dict)
|
||||
# Use canonical form
|
||||
return json.dumps(
|
||||
value, cls=JSONEncoder, separators=(',', ':'), sort_keys=True)
|
||||
value = {k: v for k, v in value.items() if v is not None}
|
||||
return dumps(value)
|
||||
|
||||
def translated(self, name=None, type_='values'):
|
||||
"Return a descriptor for the translated value of the field"
|
||||
@@ -49,6 +58,88 @@ class Dict(Field):
|
||||
raise ValueError('Missing name argument')
|
||||
return TranslatedDict(name, type_)
|
||||
|
||||
def _domain_column(self, operator, column, key=None):
|
||||
database = Transaction().database
|
||||
column = database.json_get(
|
||||
super()._domain_column(operator, column), key)
|
||||
if operator.endswith('like'):
|
||||
column = Cast(column, database.sql_type('VARCHAR').base)
|
||||
if self.search_unaccented and operator.endswith('ilike'):
|
||||
column = database.unaccent(column)
|
||||
return column
|
||||
|
||||
def _domain_value(self, operator, value):
|
||||
if backend.name() == 'sqlite' and isinstance(value, bool):
|
||||
# json_extract returns 0 for JSON false and 1 for JSON true
|
||||
value = int(value)
|
||||
if isinstance(value, (Select, CombiningQuery)):
|
||||
return value
|
||||
if operator.endswith('in'):
|
||||
return [dumps(v) for v in value]
|
||||
else:
|
||||
value = dumps(value)
|
||||
if self.search_unaccented and operator.endswith('ilike'):
|
||||
database = Transaction().database
|
||||
value = database.unaccent(value)
|
||||
return value
|
||||
|
||||
def _domain_add_null(self, column, operator, value, expression):
|
||||
expression = super()._domain_add_null(
|
||||
column, operator, value, expression)
|
||||
if value is None and operator.endswith('='):
|
||||
if operator == '=':
|
||||
expression |= (column == Null)
|
||||
else:
|
||||
expression &= (column != Null)
|
||||
return expression
|
||||
|
||||
def convert_domain(self, domain, tables, Model):
|
||||
name, operator, value = domain[:3]
|
||||
if '.' not in name:
|
||||
return super().convert_domain(domain, tables, Model)
|
||||
database = Transaction().database
|
||||
table, _ = tables[None]
|
||||
name, key = name.split('.', 1)
|
||||
Operator = SQL_OPERATORS[operator]
|
||||
column = self.sql_column(table)
|
||||
column = self._domain_column(operator, column, key)
|
||||
expression = Operator(column, self._domain_value(operator, value))
|
||||
if operator in {'=', '!='}:
|
||||
# Try to use custom operators in case there is indexes
|
||||
raw_column = self.sql_column(table)
|
||||
try:
|
||||
if value is None:
|
||||
expression = database.json_key_exists(
|
||||
raw_column, key)
|
||||
if operator == '=':
|
||||
expression = operators.Not(expression)
|
||||
return expression
|
||||
else:
|
||||
expression = database.json_contains(
|
||||
raw_column, dumps({key: value}))
|
||||
if operator == '!=':
|
||||
expression = operators.Not(expression)
|
||||
expression &= database.json_key_exists(
|
||||
raw_column, key)
|
||||
return expression
|
||||
except NotImplementedError:
|
||||
pass
|
||||
if isinstance(expression, operators.In) and not expression.right:
|
||||
expression = Literal(False)
|
||||
elif isinstance(expression, operators.NotIn) and not expression.right:
|
||||
expression = Literal(True)
|
||||
expression = self._domain_add_null(column, operator, value, expression)
|
||||
return expression
|
||||
|
||||
def convert_order(self, name, tables, Model):
|
||||
fname, _, key = name.partition('.')
|
||||
if not key:
|
||||
return super().convert_order(fname, tables, Model)
|
||||
database = Transaction().database
|
||||
table, _ = tables[None]
|
||||
column = self.sql_column(table)
|
||||
return [database.json_get(column, key)]
|
||||
|
||||
|
||||
class TranslatedDict(object):
|
||||
'A descriptor for translated values of Dict field'
|
||||
|
||||
@@ -40,6 +40,29 @@ class DictJSONB(ModelSQL):
|
||||
dico = fields.Dict('test.dict.schema', 'Test Dict')
|
||||
|
||||
|
||||
class DictNoSchema(ModelSQL):
|
||||
"Dict No Schema"
|
||||
__name__ = 'test.dict_noschema'
|
||||
dico = fields.Dict(None, "Dict")
|
||||
|
||||
|
||||
class DictUnaccentedOn(ModelSQL):
|
||||
"Dict Unaccented On"
|
||||
__name__ = 'test.dict_unaccented_on'
|
||||
dico = fields.Dict(None, "Dict")
|
||||
|
||||
|
||||
class DictUnaccentedOff(ModelSQL):
|
||||
"Dict Unaccented Off"
|
||||
__name__ = 'test.dict_unaccented_off'
|
||||
dico = fields.Dict(None, "Dict")
|
||||
|
||||
@classmethod
|
||||
def __setup__(cls):
|
||||
super().__setup__()
|
||||
cls.dico.search_unaccented = False
|
||||
|
||||
|
||||
def register(module):
|
||||
Pool.register(
|
||||
DictSchema,
|
||||
@@ -47,4 +70,7 @@ def register(module):
|
||||
DictDefault,
|
||||
DictRequired,
|
||||
DictJSONB,
|
||||
DictNoSchema,
|
||||
DictUnaccentedOn,
|
||||
DictUnaccentedOff,
|
||||
module=module, type_='model')
|
||||
|
||||
@@ -454,20 +454,15 @@ class FieldCharTranslatedTestCase(unittest.TestCase, CommonTestCaseMixin):
|
||||
return Pool().get('test.char_translate')
|
||||
|
||||
|
||||
@unittest.skipUnless(backend.name() == 'postgresql',
|
||||
"unaccent works only on postgresql")
|
||||
class FieldCharUnaccentedTestCase(unittest.TestCase):
|
||||
"Test Field Char with unaccented searches"
|
||||
|
||||
class UnaccentedTestCase(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super(FieldCharUnaccentedTestCase, cls).setUpClass()
|
||||
activate_module('tests')
|
||||
super().setUpClass()
|
||||
cls._activate_extension()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
super(FieldCharUnaccentedTestCase, cls).tearDownClass()
|
||||
super().tearDownClass()
|
||||
cls._deactivate_extension()
|
||||
|
||||
@classmethod
|
||||
@@ -493,6 +488,17 @@ class FieldCharUnaccentedTestCase(unittest.TestCase):
|
||||
Database = backend.get('Database')
|
||||
Database._has_unaccent.clear()
|
||||
|
||||
|
||||
@unittest.skipUnless(backend.name() == 'postgresql',
|
||||
"unaccent works only on postgresql")
|
||||
class FieldCharUnaccentedTestCase(UnaccentedTestCase):
|
||||
"Test Field Char with unaccented searches"
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
activate_module('tests')
|
||||
super().setUpClass()
|
||||
|
||||
@with_transaction()
|
||||
def test_normal_search(self):
|
||||
"Test searches without the unaccented feature"
|
||||
|
||||
@@ -8,6 +8,8 @@ from trytond.pool import Pool
|
||||
from trytond.tests.test_tryton import activate_module, with_transaction
|
||||
from trytond.transaction import Transaction
|
||||
|
||||
from .test_field_char import UnaccentedTestCase
|
||||
|
||||
|
||||
class FieldDictTestCase(unittest.TestCase):
|
||||
"Test Field Dict"
|
||||
@@ -129,7 +131,7 @@ class FieldDictTestCase(unittest.TestCase):
|
||||
|
||||
@with_transaction()
|
||||
@unittest.skipIf(
|
||||
backend.name() != 'postgresql', 'jsonb only suported by postgresql')
|
||||
backend.name() != 'postgresql', 'jsonb only supported by postgresql')
|
||||
def test_create_jsonb(self):
|
||||
"Test create dict as jsonb"
|
||||
connection = Transaction().connection
|
||||
@@ -177,7 +179,7 @@ class FieldDictTestCase(unittest.TestCase):
|
||||
|
||||
@with_transaction()
|
||||
@unittest.skipIf(
|
||||
backend.name() != 'postgresql', 'jsonb only suported by postgresql')
|
||||
backend.name() != 'postgresql', 'jsonb only supported by postgresql')
|
||||
def test_write_jsonb(self):
|
||||
"Test write dict as jsonb"
|
||||
connection = Transaction().connection
|
||||
@@ -195,6 +197,518 @@ class FieldDictTestCase(unittest.TestCase):
|
||||
|
||||
self.assertDictEqual(dict_.dico, {'z': 26})
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_equals(self):
|
||||
"Test search dict element equals"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 'Foo'},
|
||||
}])
|
||||
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', '=', "Foo"),
|
||||
])
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', '=', "Bar"),
|
||||
])
|
||||
dicts_foo_b = Dict.search([
|
||||
('dico.b', '=', "Foo"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_foo, [dict_])
|
||||
self.assertListEqual(dicts_bar, [])
|
||||
self.assertListEqual(dicts_foo_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_equals_none(self):
|
||||
"Test search dict element equals None"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': None},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '=', None),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_non_element_equals_none(self):
|
||||
"Test search dict non element equals None"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Foo"},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.b', '=', None),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_non_equals(self):
|
||||
"Test search dict element non equals"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Foo"},
|
||||
}])
|
||||
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', '!=', "Foo"),
|
||||
])
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', '!=', "Bar"),
|
||||
])
|
||||
dicts_foo_b = Dict.search([
|
||||
('dico.b', '!=', "Foo"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_foo, [])
|
||||
self.assertListEqual(dicts_bar, [dict_])
|
||||
self.assertListEqual(dicts_foo_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_non_equals_none(self):
|
||||
"Test search dict element non equals None"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': None},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '!=', None),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_non_element_non_equals_none(self):
|
||||
"Test search dict non element non equals None"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Foo"},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.b', '!=', None),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_equals_true(self):
|
||||
"Test search dict element equals True"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict1, dict2 = Dict.create([{
|
||||
'dico': {'a': True},
|
||||
}, {
|
||||
'dico': {'a': False},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '=', True),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict1])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_equals_false(self):
|
||||
"Test search dict element equals False"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict1, dict2 = Dict.create([{
|
||||
'dico': {'a': True},
|
||||
}, {
|
||||
'dico': {'a': False},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '=', False),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict2])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_non_equals_true(self):
|
||||
"Test search dict element non equals True"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': True},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '!=', True),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_in(self):
|
||||
"Test search dict element in"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Foo"},
|
||||
}])
|
||||
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'in', ["Foo"]),
|
||||
])
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'in', ["Bar"]),
|
||||
])
|
||||
dicts_empty = Dict.search([
|
||||
('dico.a', 'in', []),
|
||||
])
|
||||
dicts_foo_b = Dict.search([
|
||||
('dico.b', 'in', ["Foo"]),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_foo, [dict_])
|
||||
self.assertListEqual(dicts_bar, [])
|
||||
self.assertListEqual(dicts_empty, [])
|
||||
self.assertListEqual(dicts_foo_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_in_none(self):
|
||||
"Test search dict element in [None]"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': None},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', 'in', [None]),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_not_in(self):
|
||||
"Test search dict element not in"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Foo"},
|
||||
}])
|
||||
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'not in', ["Foo"]),
|
||||
])
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'not in', ["Bar"]),
|
||||
])
|
||||
dicts_empty = Dict.search([
|
||||
('dico.a', 'not in', []),
|
||||
])
|
||||
dicts_foo_b = Dict.search([
|
||||
('dico.b', 'not in', ["Foo"]),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_foo, [])
|
||||
self.assertListEqual(dicts_bar, [dict_])
|
||||
self.assertListEqual(dicts_empty, [dict_])
|
||||
self.assertListEqual(dicts_foo_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_not_in_none(self):
|
||||
"Test search dict element not in [None]"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': None},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', 'not in', [None]),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_less(self):
|
||||
"Test search dict element less than"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 1.1},
|
||||
}])
|
||||
|
||||
dicts_5 = Dict.search([
|
||||
('dico.a', '<', 5),
|
||||
])
|
||||
dicts__5 = Dict.search([
|
||||
('dico.a', '<', -5),
|
||||
])
|
||||
dicts_1_1 = Dict.search([
|
||||
('dico.a', '<', 1.1),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_5, [dict_])
|
||||
self.assertListEqual(dicts__5, [])
|
||||
self.assertListEqual(dicts_1_1, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_less_equals(self):
|
||||
"Test search dict element less than or equals"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 1.1},
|
||||
}])
|
||||
|
||||
dicts_5 = Dict.search([
|
||||
('dico.a', '<=', 5),
|
||||
])
|
||||
dicts__5 = Dict.search([
|
||||
('dico.a', '<=', -5),
|
||||
])
|
||||
dicts_1_1 = Dict.search([
|
||||
('dico.a', '<=', 1.1),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_5, [dict_])
|
||||
self.assertListEqual(dicts__5, [])
|
||||
self.assertListEqual(dicts_1_1, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_greater(self):
|
||||
"Test search dict element greater than"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 1.1},
|
||||
}])
|
||||
|
||||
dicts_5 = Dict.search([
|
||||
('dico.a', '>', 5),
|
||||
])
|
||||
dicts__5 = Dict.search([
|
||||
('dico.a', '>', -5),
|
||||
])
|
||||
dicts_1_1 = Dict.search([
|
||||
('dico.a', '>', 1.1),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_5, [])
|
||||
self.assertListEqual(dicts__5, [dict_])
|
||||
self.assertListEqual(dicts_1_1, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_greater_equals(self):
|
||||
"Test search dict element greater than or equals"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 1.1},
|
||||
}])
|
||||
|
||||
dicts_5 = Dict.search([
|
||||
('dico.a', '>=', 5),
|
||||
])
|
||||
dicts__5 = Dict.search([
|
||||
('dico.a', '>=', -5),
|
||||
])
|
||||
dicts_1_1 = Dict.search([
|
||||
('dico.a', '>=', 1.1),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_5, [])
|
||||
self.assertListEqual(dicts__5, [dict_])
|
||||
self.assertListEqual(dicts_1_1, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_like(self):
|
||||
"Test search dict element like"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Bar"},
|
||||
}])
|
||||
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'like', "Bar"),
|
||||
])
|
||||
dicts_b = Dict.search([
|
||||
('dico.a', 'like', "B%"),
|
||||
])
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'like', "Foo"),
|
||||
])
|
||||
dicts_f = Dict.search([
|
||||
('dico.a', 'like', "F%"),
|
||||
])
|
||||
dicts_b_b = Dict.search([
|
||||
('dico.b', 'like', "B%"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_bar, [dict_])
|
||||
self.assertListEqual(dicts_b, [dict_])
|
||||
self.assertListEqual(dicts_foo, [])
|
||||
self.assertListEqual(dicts_f, [])
|
||||
self.assertListEqual(dicts_b_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_ilike(self):
|
||||
"Test search dict element ilike"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Bar"},
|
||||
}])
|
||||
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'ilike', "bar"),
|
||||
])
|
||||
dicts_b = Dict.search([
|
||||
('dico.a', 'ilike', "b%"),
|
||||
])
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'ilike', "foo"),
|
||||
])
|
||||
dicts_f = Dict.search([
|
||||
('dico.a', 'ilike', "f%"),
|
||||
])
|
||||
dicts_b_b = Dict.search([
|
||||
('dico.b', 'ilike', "b%"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_bar, [dict_])
|
||||
self.assertListEqual(dicts_b, [dict_])
|
||||
self.assertListEqual(dicts_foo, [])
|
||||
self.assertListEqual(dicts_f, [])
|
||||
self.assertListEqual(dicts_b_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_not_like(self):
|
||||
"Test search dict element not like"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Bar"},
|
||||
}])
|
||||
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'not like', "Bar"),
|
||||
])
|
||||
dicts_b = Dict.search([
|
||||
('dico.a', 'not like', "B%"),
|
||||
])
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'not like', "Foo"),
|
||||
])
|
||||
dicts_f = Dict.search([
|
||||
('dico.a', 'not like', "F%"),
|
||||
])
|
||||
dicts_b_b = Dict.search([
|
||||
('dico.b', 'not like', "B%"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_bar, [])
|
||||
self.assertListEqual(dicts_b, [])
|
||||
self.assertListEqual(dicts_foo, [dict_])
|
||||
self.assertListEqual(dicts_f, [dict_])
|
||||
self.assertListEqual(dicts_b_b, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_element_not_ilike(self):
|
||||
"Test search dict element not ilike"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': "Bar"},
|
||||
}])
|
||||
|
||||
dicts_bar = Dict.search([
|
||||
('dico.a', 'not ilike', "bar"),
|
||||
])
|
||||
dicts_b = Dict.search([
|
||||
('dico.a', 'not ilike', "b%"),
|
||||
])
|
||||
dicts_foo = Dict.search([
|
||||
('dico.a', 'not ilike', "foo"),
|
||||
])
|
||||
dicts_f = Dict.search([
|
||||
('dico.a', 'not ilike', "f%"),
|
||||
])
|
||||
dicts_b_b = Dict.search([
|
||||
('dico.b', 'not ilike', "b%"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts_bar, [])
|
||||
self.assertListEqual(dicts_b, [])
|
||||
self.assertListEqual(dicts_foo, [dict_])
|
||||
self.assertListEqual(dicts_f, [dict_])
|
||||
self.assertListEqual(dicts_b_b, [])
|
||||
|
||||
@with_transaction()
|
||||
@unittest.skipIf(
|
||||
backend.name() != 'postgresql', 'jsonb only supported by postgresql')
|
||||
def test_search_element_jsonb(self):
|
||||
"Test search dict element on jsonb"
|
||||
connection = Transaction().connection
|
||||
Database = backend.get('Database')
|
||||
if Database().get_version(connection) < (9, 2):
|
||||
return
|
||||
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
self.set_jsonb(Dict._table)
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 'Foo'},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', '=', "Foo"),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_order_element(self):
|
||||
"Test search order by dict element"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_noschema')
|
||||
for value in [5, 3, 2]:
|
||||
Dict.create([{'dico': {'a': 5 - value, 'b': value}}])
|
||||
|
||||
records = Dict.search([], order=[('dico.b', 'ASC')])
|
||||
values = [r.dico['b'] for r in records]
|
||||
|
||||
self.assertListEqual(values, [2, 3, 5])
|
||||
|
||||
@with_transaction()
|
||||
def test_string(self):
|
||||
"Test string dict"
|
||||
@@ -221,5 +735,61 @@ class FieldDictTestCase(unittest.TestCase):
|
||||
dict_.dico_string_keys, {'a': 'A', 'type': "Type"})
|
||||
|
||||
|
||||
@unittest.skipUnless(backend.name() == 'postgresql',
|
||||
"unaccent works only on postgresql")
|
||||
class FieldDictUnaccentedTestCase(UnaccentedTestCase):
|
||||
"Test Field Dict with unaccented searched"
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
activate_module('tests')
|
||||
super().setUpClass()
|
||||
|
||||
@with_transaction()
|
||||
def test_search_unaccented_off(self):
|
||||
"Test searches without the unaccented feature"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_unaccented_off')
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 'Stéphanie'},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', 'ilike', 'Stephanie'),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_accented(self):
|
||||
"Test searches of accented value"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_unaccented_on')
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 'Stéphanie'},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', 'ilike', 'Stephanie'),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
@with_transaction()
|
||||
def test_search_unaccented(self):
|
||||
"Test searches of unaccented value"
|
||||
pool = Pool()
|
||||
Dict = pool.get('test.dict_unaccented_on')
|
||||
dict_, = Dict.create([{
|
||||
'dico': {'a': 'Stephanie'},
|
||||
}])
|
||||
|
||||
dicts = Dict.search([
|
||||
('dico.a', 'ilike', 'Stéphanie'),
|
||||
])
|
||||
|
||||
self.assertListEqual(dicts, [dict_])
|
||||
|
||||
|
||||
def suite():
|
||||
return unittest.TestLoader().loadTestsFromTestCase(FieldDictTestCase)
|
||||
|
||||
Reference in New Issue
Block a user