1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
# 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 collections import OrderedDict
from functools import wraps
class Workflow(object):
"""
Mix-in class to handle transition check.
"""
__slots__ = ()
_transition_state = 'state'
@classmethod
def __setup__(cls):
super(Workflow, cls).__setup__()
cls._transitions = set()
@staticmethod
def transition(state):
def check_transition(func):
@wraps(func)
def wrapper(cls, records, *args, **kwargs):
filtered = []
to_update = OrderedDict()
assert len(records) == len(set(records)), "Duplicate records"
for record in records:
current_state = getattr(record, cls._transition_state)
transition = (current_state, state)
if transition in cls._transitions:
filtered.append(record)
if current_state != state:
to_update[record] = current_state
result = func(cls, filtered, *args, **kwargs)
if to_update:
for record in list(to_update.keys()):
current_state = getattr(record, cls._transition_state)
if current_state != to_update[record]:
del to_update[record]
to_update = list(to_update)
cls.write(to_update, {
cls._transition_state: state,
})
cls.log(
to_update, 'transition',
f'{cls._transition_state}:{state}')
return result
return wrapper
return check_transition
@classmethod
def copy(cls, records, default=None):
if default is None:
default = {}
else:
default = default.copy()
default.setdefault(
cls._transition_state, cls._defaults[cls._transition_state]())
return super().copy(records, default=default)
|
