Skip to content

API

reversible_function

Bases: t.Generic[_P, _Q, _R]

Source code in carnot/reversible.py 
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class reversible_function(t.Generic[_P, _Q, _R]):

    def __init__(
        self,
        f_forward: t.Callable[_P, _R],
        f_backward: t.Optional[t.Callable[_Q, None]] = None,
    ) -> None:
        self.__doc__ = f_forward.__doc__
        self._f_forward = f_forward
        self._f_backward = f_backward
        self._args: t.Tuple[t.Any, ...] = ()

    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R:
        return_value = self._f_forward(*args, **kwargs)

        if self._f_backward is not None:
            for record in inspect.stack():
                if "__reversible_stack__" in record.frame.f_locals:
                    stack = record.frame.f_locals["__reversible_stack__"]
                    stack.append((self._f_backward, self._args))
                    break

        return return_value

    def backward(
        self,
        f_backward: t.Callable[_Q, None],
    ) -> reversible_function:
        """Registers a function for backward process.

        Args:
            f_backward: Function for backward process.

        Returns:
            `reversible_function` itself.
        """
        self._f_backward = f_backward
        return self

    def set_args(self, *args: t.Any) -> None:
        """Sets arguments for the backward function.

        Args:
            *args: Arguments given to the backward function.
        """
        self._args = args

    def get_forward(self) -> t.Callable[_P, _R]:
        """Returns the forward function.

        Returns:
            Registered forward function.
        """
        return self._f_forward

    def get_backward(self) -> t.Callable[_Q, None]:
        """Returns the backward function.

        Returns:
            Registered backward function.
        """
        return self._f_backward

backward(f_backward)

Registers a function for backward process.

Parameters:

Name Type Description Default
f_backward t.Callable[_Q, None]

Function for backward process.

 required

Returns:

Type Description
reversible_function

reversible_function itself.
Source code in carnot/reversible.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def backward(
    self,
    f_backward: t.Callable[_Q, None],
) -> reversible_function:
    """Registers a function for backward process.

    Args:
        f_backward: Function for backward process.

    Returns:
        `reversible_function` itself.
    """
    self._f_backward = f_backward
    return self

get_backward()

Returns the backward function.

Returns:

Type Description
t.Callable[_Q, None]

Registered backward function.
Source code in carnot/reversible.py 
72
73
74
75
76
77
78
def get_backward(self) -> t.Callable[_Q, None]:
    """Returns the backward function.

    Returns:
        Registered backward function.
    """
    return self._f_backward

get_forward()

Returns the forward function.

Returns:

Type Description
t.Callable[_P, _R]

Registered forward function.
Source code in carnot/reversible.py 
64
65
66
67
68
69
70
def get_forward(self) -> t.Callable[_P, _R]:
    """Returns the forward function.

    Returns:
        Registered forward function.
    """
    return self._f_forward

set_args(*args)

Sets arguments for the backward function.

Parameters:

Name Type Description Default
*args t.Any

Arguments given to the backward function.

 ()
Source code in carnot/reversible.py 
56
57
58
59
60
61
62
def set_args(self, *args: t.Any) -> None:
    """Sets arguments for the backward function.

    Args:
        *args: Arguments given to the backward function.
    """
    self._args = args

reversible_method

Bases: t.Generic[_Object_t, _P, _Q, _R]

Descriptor for reversible methods.

This descriptor can be applied to reversible methods, whose processes are not unidirectional. Using this descriptor, developers can define the forward process and backward process, the former is processed first in a transaction and the latter is processed if the transaction failes.

Transactions in reversible are defined using reversible.transaction. For more information about transactions, see the documentation.

Source code in carnot/reversible.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class reversible_method(t.Generic[_Object_t, _P, _Q, _R]):
    """Descriptor for reversible methods.

    This descriptor can be applied to reversible methods, whose processes are
    not unidirectional. Using this descriptor, developers can define the
    forward process and backward process, the former is processed first in a
    transaction and the latter is processed if the transaction failes.

    Transactions in `reversible` are defined using `reversible.transaction`.
    For more information about transactions, see the documentation.
    """

    def __init__(
        self,
        f_forward: t.Callable[Concatenate[_Object_t, _P], _R],
        f_backward: t.Optional[
            t.Callable[Concatenate[_Object_t, _Q], None]
        ] = None,
    ) -> None:
        """
        Args:
            f_forward: Function for forward process.
            f_backward: Function for backward process.
        """
        self.__doc__ = f_forward.__doc__
        self._f_forward = f_forward
        self._f_backward = f_backward
        self._args: t.Tuple[t.Any, ...] = ()

    @t.overload
    def __get__(
        self,
        instance: None,
        owner: t.Optional[t.Type[_Object_t]] = None,
    ) -> reversible_method:
        ...

    @t.overload
    def __get__(
        self,
        instance: _Object_t,
        owner: t.Optional[t.Type[_Object_t]] = None,
    ) -> t.Callable[_P, _R]:
        ...

    def __get__(self, instance, owner = None):
        if instance is None:
            return self

        @functools.wraps(self._f_forward)
        def _callback(*args, **kwargs):
            return_value = self._f_forward(instance, *args, **kwargs)

            if self._f_backward is not None:
                for record in inspect.stack():
                    if "__reversible_stack__" in record.frame.f_locals:
                        stack = record.frame.f_locals["__reversible_stack__"]
                        stack.append((self._f_backward, self._args))
                        break

            return return_value

        return _callback

    def backward(
        self,
        f_backward: t.Callable[Concatenate[_Object_t, _Q], None],
    ) -> reversible_method:
        """Registers a function for backward process.

        Args:
            f_backward: Function for backward process.

        Returns:
            `reversible_method` itself.
        """
        self._f_backward = f_backward
        return self

    def set_args(self, *args: t.Any) -> None:
        """Sets arguments for the backward function.

        Args:
            *args: Arguments given to the backward function.
        """
        self._args = args

    def get_forward(self) -> t.Callable[[Concatenate[_Object_t, _P], _R]]:
        """Returns the forward function.

        Returns:
            Registered forward function.
        """
        return self._f_forward

    def get_backward(self) -> t.Callable[[Concatenate[_Object_t, _Q], None]]:
        """Returns the backward function.

        Returns:
            Registered backward function.
        """
        return self._f_backward

__init__(f_forward, f_backward=None)

Parameters:

Name Type Description Default
f_forward t.Callable[Concatenate[_Object_t, _P], _R]

Function for forward process.

 required
f_backward t.Optional[t.Callable[Concatenate[_Object_t, _Q], None]]

Function for backward process.

 None
Source code in carnot/reversible.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def __init__(
    self,
    f_forward: t.Callable[Concatenate[_Object_t, _P], _R],
    f_backward: t.Optional[
        t.Callable[Concatenate[_Object_t, _Q], None]
    ] = None,
) -> None:
    """
    Args:
        f_forward: Function for forward process.
        f_backward: Function for backward process.
    """
    self.__doc__ = f_forward.__doc__
    self._f_forward = f_forward
    self._f_backward = f_backward
    self._args: t.Tuple[t.Any, ...] = ()

backward(f_backward)

Registers a function for backward process.

Parameters:

Name Type Description Default
f_backward t.Callable[Concatenate[_Object_t, _Q], None]

Function for backward process.

 required

Returns:

Type Description
reversible_method

reversible_method itself.
Source code in carnot/reversible.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def backward(
    self,
    f_backward: t.Callable[Concatenate[_Object_t, _Q], None],
) -> reversible_method:
    """Registers a function for backward process.

    Args:
        f_backward: Function for backward process.

    Returns:
        `reversible_method` itself.
    """
    self._f_backward = f_backward
    return self

get_backward()

Returns the backward function.

Returns:

Type Description
t.Callable[[Concatenate[_Object_t, _Q], None]]

Registered backward function.
Source code in carnot/reversible.py 
179
180
181
182
183
184
185
def get_backward(self) -> t.Callable[[Concatenate[_Object_t, _Q], None]]:
    """Returns the backward function.

    Returns:
        Registered backward function.
    """
    return self._f_backward

get_forward()

Returns the forward function.

Returns:

Type Description
t.Callable[[Concatenate[_Object_t, _P], _R]]

Registered forward function.
Source code in carnot/reversible.py 
171
172
173
174
175
176
177
def get_forward(self) -> t.Callable[[Concatenate[_Object_t, _P], _R]]:
    """Returns the forward function.

    Returns:
        Registered forward function.
    """
    return self._f_forward

set_args(*args)

Sets arguments for the backward function.

Parameters:

Name Type Description Default
*args t.Any

Arguments given to the backward function.

 ()
Source code in carnot/reversible.py 
163
164
165
166
167
168
169
def set_args(self, *args: t.Any) -> None:
    """Sets arguments for the backward function.

    Args:
        *args: Arguments given to the backward function.
    """
    self._args = args

transaction(callable)

Wrapper to define a transaction where some reversible process can be processed.

This decorator tracks reversible methods in given callable and execute the backward processes if any errors occur in the callable.

Parameters:

Name Type Description Default
callable _Callable_t

Callable to be wrapped.

 required

Returns:

Type Description
_Callable_t

Wrapped callable.
Source code in carnot/transaction.py 
 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
def transaction(callable: _Callable_t) -> _Callable_t:
    """Wrapper to define a transaction where some reversible process can be
    processed.

    This decorator tracks reversible methods in given callable and execute
    the backward processes if any errors occur in the callable.

    Args:
        callable: Callable to be wrapped.

    Returns:
        Wrapped callable.
    """
    stack: t.Deque[t.Tuple[t.Callable, t.Tuple[t.Any, ...]]] = deque()

    @functools.wraps(callable)
    def _callable(*args, **kwargs):
        __reversible_stack__ = stack

        try:
            return_value = callable(*args, **kwargs)
        except Exception as e:
            while len(stack):
                backward, args_ = stack.pop()
                backward(*args_)
            raise e
        else:
            return return_value

    return _callable