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

Add Exclude constraint

Browse Source 
issue1422
review44061002
This commit is contained in:
Cédric Krier
2018-03-19 16:57:49 +01:00
parent acdbe5a08f
commit b3ca13748b
9 changed files with 242 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG
+1
View File
@@ -1,3 +1,4 @@
* Add Exclude constraint
* Allow to perform unaccented searches on Char fields
* Always raise exception in table handler
* Simplify API for Session: new, remove, check and reset
doc/ref/models/models.rst
+31
View File
@@ -527,6 +527,37 @@ Instance attributes:
The tuple of SQL Column instances.
.. attribute:: Unique.operators
The tuple of `Equal` operators.
Exclude
-------
.. class:: Exclude(table[, (expression, operator), ...[, where]])
It represents an exclude :class:`Constraint` which guarantees that if any two
rows are compared on the specified expression using the specified operator not
all of these comparisons will return `TRUE`.
Instance attributes:
.. attribute:: Exclude.excludes
The tuple of expression and operator.
.. attribute:: Exclude.columns
The tuple of expressions.
.. attribute:: Exclude.operators
The tuple of operators.
.. attribute:: Exclude.where
The clause for which the exclusion applies.
========
Workflow
========
trytond/backend/database.py
+1 -1
View File
@@ -135,7 +135,7 @@ class DatabaseInterface(object):
'''
raise NotImplementedError
def has_constraint(self):
def has_constraint(self, constraint):
'''
Return True if database handle constraint.
trytond/backend/postgresql/database.py
+1 -1
View File
@@ -265,7 +265,7 @@ class Database(DatabaseInterface):
cursor = connection.cursor()
cursor.execute('LOCK "%s" IN EXCLUSIVE MODE NOWAIT' % table)
def has_constraint(self):
def has_constraint(self, constraint):
return True
def has_multirow_insert(self):
trytond/backend/sqlite/database.py
+1 -1
View File
@@ -410,7 +410,7 @@ class Database(DatabaseInterface):
def lock(self, connection, table):
pass
def has_constraint(self):
def has_constraint(self, constraint):
return False
def has_multirow_insert(self):
trytond/model/__init__.py
+2 -2
View File
@@ -4,7 +4,7 @@ from .model import Model
from .modelview import ModelView
from .modelstorage import ModelStorage, EvalEnvironment
from .modelsingleton import ModelSingleton
from .modelsql import ModelSQL, Check, Unique
from .modelsql import ModelSQL, Check, Unique, Exclude
from .workflow import Workflow
from .dictschema import DictSchemaMixin
from .match import MatchMixin
@@ -14,7 +14,7 @@ from .descriptors import dualmethod
from .order import sequence_ordered
__all__ = ['Model', 'ModelView', 'ModelStorage', 'ModelSingleton', 'ModelSQL',
'Check', 'Unique',
'Check', 'Unique', 'Exclude',
'Workflow', 'DictSchemaMixin', 'MatchMixin', 'UnionMixin', 'dualmethod',
'MultiValueMixin', 'ValueMixin',
'EvalEnvironment', 'sequence_ordered']
trytond/model/modelsql.py
+68 -9
View File
@@ -8,7 +8,7 @@ from sql import (Table, Column, Literal, Desc, Asc, Expression, Null,
NullsFirst, NullsLast)
from sql.functions import CurrentTimestamp, Extract
from sql.conditionals import Coalesce
from sql.operators import Or, And, Operator
from sql.operators import Or, And, Operator, Equal
from sql.aggregate import Count, Max
from trytond.model import ModelStorage, ModelView
@@ -77,6 +77,10 @@ class Unique(Constraint):
def columns(self):
return self._columns
@property
def operators(self):
return tuple(Equal for c in self._columns)
def __str__(self):
return 'UNIQUE(%s)' % (', '.join(map(str, self.columns)))
@@ -88,6 +92,53 @@ class Unique(Constraint):
return tuple(p)
class Exclude(Constraint):
__slots__ = ('_excludes', '_where')
def __init__(self, table, *excludes, **kwargs):
super(Exclude, self).__init__(table)
assert all(isinstance(c, Expression) and issubclass(o, Operator)
for c, o in excludes), excludes
self._excludes = tuple(excludes)
where = kwargs.get('where')
if where is not None:
assert isinstance(where, Expression)
self._where = where
@property
def excludes(self):
return self._excludes
@property
def columns(self):
return tuple(c for c, _ in self._excludes)
@property
def operators(self):
return tuple(o for _, o in self._excludes)
@property
def where(self):
return self._where
def __str__(self):
exclude = ', '.join('%s WITH %s' % (column, operator._operator)
for column, operator in self.excludes)
where = ''
if self.where:
where = ' WHERE ' + str(self.where)
return 'EXCLUDE (%s)' % exclude + where
@property
def params(self):
p = []
for column, operator in self._excludes:
p.extend(column.params)
if self.where:
p.extend(self.where.params)
return tuple(p)
class ModelSQL(ModelStorage):
"""
Define a model with storage in database.
@@ -1451,30 +1502,38 @@ class ModelSQL(ModelStorage):
def validate(cls, records):
super(ModelSQL, cls).validate(records)
transaction = Transaction()
database = transaction.database
connection = transaction.connection
has_constraint = database.has_constraint
lock = database.lock
cursor = transaction.connection.cursor()
if transaction.database.has_constraint():
return
# Works only for a single transaction
ids = map(int, records)
for _, sql, error in cls._sql_constraints:
if has_constraint(sql):
continue
table = sql.table
if isinstance(sql, Unique):
columns = [Column(table, c.name) for c in sql.columns]
if isinstance(sql, (Unique, Exclude)):
lock(connection, cls._table)
columns = list(sql.columns)
columns.insert(0, table.id)
in_max = transaction.database.IN_MAX // (len(columns) + 1)
for sub_ids in grouped_slice(ids, in_max):
red_sql = reduce_ids(table.id, sub_ids)
where = reduce_ids(table.id, sub_ids)
if isinstance(sql, Exclude):
where &= sql.where
cursor.execute(*table.select(*columns, where=red_sql))
cursor.execute(*table.select(*columns, where=where))
where = Literal(False)
for row in cursor.fetchall():
clause = table.id != row[0]
for column, value in zip(sql.columns, row[1:]):
for column, operator, value in zip(
sql.columns, sql.operators, row[1:]):
if value is None:
# NULL is always unique
clause &= Literal(False)
clause &= Column(table, column.name) == value
clause &= operator(column, value)
where |= clause
cursor.execute(
*table.select(table.id, where=where, limit=1))
trytond/tests/modelsql.py
+50 -1
View File
@@ -1,6 +1,8 @@
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
from trytond.model import ModelSQL, fields
from sql.operators import Equal
from trytond.model import ModelSQL, fields, Check, Unique, Exclude
from trytond.pool import Pool
@@ -58,6 +60,50 @@ class ModelTranslation(ModelSQL):
name = fields.Char("Name", translate=True)
class ModelCheck(ModelSQL):
"ModelSQL with check constraint"
__name__ = 'test.modelsql.check'
value = fields.Integer("Value")
@classmethod
def __setup__(cls):
super(ModelCheck, cls).__setup__()
t = cls.__table__()
cls._sql_constraints = [
('check', Check(t, (t.value > 42)),
"Value must be greater than 42."),
]
class ModelUnique(ModelSQL):
"ModelSQL with unique constraint"
__name__ = 'test.modelsql.unique'
value = fields.Integer("Value")
@classmethod
def __setup__(cls):
super(ModelUnique, cls).__setup__()
t = cls.__table__()
cls._sql_constraints = [
('unique', Unique(t, t.value), "Value must be unique."),
]
class ModelExclude(ModelSQL):
"ModelSQL with exclude constraint"
__name__ = 'test.modelsql.exclude'
value = fields.Integer("Value")
@classmethod
def __setup__(cls):
super(ModelExclude, cls).__setup__()
t = cls.__table__()
cls._sql_constraints = [
('exclude', Exclude(t, (t.value, Equal), where=t.value > 0),
"Value must be unique."),
]
def register(module):
Pool.register(
ModelSQLRequiredField,
@@ -67,4 +113,7 @@ def register(module):
ModelSQLOne2ManyTarget,
NullOrder,
ModelTranslation,
ModelCheck,
ModelUnique,
ModelExclude,
module=module, type_='model')
trytond/tests/test_modelsql.py
+87
View File
@@ -173,6 +173,93 @@ class ModelSQLTestCase(unittest.TestCase):
delete_ids.assert_called_with(
'test.modelsql.translation', 'model', [record.id])
@with_transaction()
def test_constraint_check(self):
"Test check constraint"
pool = Pool()
Model = pool.get('test.modelsql.check')
record, = Model.create([{'value': 50}])
self.assertTrue(record.id)
@with_transaction()
def test_constraint_check_null(self):
"Test check constraint with null"
pool = Pool()
Model = pool.get('test.modelsql.check')
record, = Model.create([{'value': None}])
self.assertTrue(record.id)
@with_transaction()
def test_constraint_check_invalid(self):
"Test invalid check constraint"
pool = Pool()
Model = pool.get('test.modelsql.check')
with self.assertRaises(UserError):
Model.create([{'value': 10}])
@with_transaction()
def test_constraint_unique(self):
"Test unique constraint"
pool = Pool()
Model = pool.get('test.modelsql.unique')
records = Model.create([{'value': 1}, {'value': 2}])
self.assertEqual(len(records), 2)
@with_transaction()
def test_constraint_unique_null(self):
"Test unique constraint with null"
pool = Pool()
Model = pool.get('test.modelsql.unique')
records = Model.create([{'value': None}, {'value': None}])
self.assertEqual(len(records), 2)
@with_transaction()
def test_constraint_unique_invalid(self):
"Test invalid unique constraint"
pool = Pool()
Model = pool.get('test.modelsql.unique')
with self.assertRaises(UserError):
Model.create([{'value': 42}, {'value': 42}])
@with_transaction()
def test_constraint_exclude(self):
"Test exclude constraint"
pool = Pool()
Model = pool.get('test.modelsql.exclude')
records = Model.create([{'value': 1}, {'value': 2}])
self.assertEqual(len(records), 2)
@with_transaction()
def test_constraint_exclude_exclusion(self):
"Test exclude constraint exclusion"
pool = Pool()
Model = pool.get('test.modelsql.exclude')
records = Model.create([{'value': -1}, {'value': -1}])
self.assertEqual(len(records), 2)
@with_transaction()
def test_constraint_exclude_invalid(self):
"Test invalid exclude constraint"
pool = Pool()
Model = pool.get('test.modelsql.exclude')
with self.assertRaises(UserError):
Model.create([{'value': 42}, {'value': 42}])
def suite():
return unittest.TestLoader().loadTestsFromTestCase(ModelSQLTestCase)