tarantool-python/tarantool/types.py at master · ThCompiler/tarantool-python · GitHub

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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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
"""
Additional Tarantool type definitions.
"""

import typing
from dataclasses import dataclass


@dataclass
class BoxError():
    """
    Type representing Tarantool `box.error`_ object: a single
    MP_ERROR_STACK object with a link to the previous stack error.

    .. _box.error: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_error/error/
    """
    # pylint: disable=too-many-instance-attributes

    type: typing.Union[str, bytes]
    """
    Type that implies source, for example ``"ClientError"``.

    Value type depends on :class:`~tarantool.Connection`
    :paramref:`~tarantool.Connection.params.encoding`.
    """

    file: typing.Union[str, bytes]
    """
    Source code file where error was caught.

    Value type depends on :class:`~tarantool.Connection`
    :paramref:`~tarantool.Connection.params.encoding`.
    """

    line: int
    """
    Line number in source code file.
    """

    message: typing.Union[str, bytes]
    """
    Text of reason.

    Value type depends on :class:`~tarantool.Connection`
    :paramref:`~tarantool.Connection.params.encoding`.
    """

    errno: int
    """
    Ordinal number of the error.
    """

    errcode: int
    """
    Number of the error as defined in ``errcode.h``.
    """

    fields: typing.Optional[dict] = None
    """
    Additional fields depending on error type. For example, if
    :attr:`~tarantool.BoxError.type` is ``"AccessDeniedError"``,
    then it will include ``"object_type"``, ``"object_name"``,
    ``"access_type"``.
    """

    prev: typing.Optional[typing.List['BoxError']] = None
    """
    Previous error in stack.
    """


MP_ERROR_STACK = 0x00
MP_ERROR_TYPE = 0x00
MP_ERROR_FILE = 0x01
MP_ERROR_LINE = 0x02
MP_ERROR_MESSAGE = 0x03
MP_ERROR_ERRNO = 0x04
MP_ERROR_ERRCODE = 0x05
MP_ERROR_FIELDS = 0x06


def decode_box_error(err_map):
    """
    Decode MessagePack map received from Tarantool to `box.error`_
    object representation.

    :param err_map: Error MessagePack map received from Tarantool.
    :type err_map: :obj:`dict`

    :rtype: :class:`~tarantool.BoxError`

    :raises: :exc:`KeyError`
    """

    encoded_stack = err_map[MP_ERROR_STACK]

    prev = None
    for item in encoded_stack[::-1]:
        err = BoxError(
            type=item[MP_ERROR_TYPE],
            file=item[MP_ERROR_FILE],
            line=item[MP_ERROR_LINE],
            message=item[MP_ERROR_MESSAGE],
            errno=item[MP_ERROR_ERRNO],
            errcode=item[MP_ERROR_ERRCODE],
            fields=item.get(MP_ERROR_FIELDS),  # omitted if empty
            prev=prev,
        )
        prev = err

    return prev


def encode_box_error(err):
    """
    Encode Python `box.error`_ representation to MessagePack map.

    :param err: Error to encode
    :type err: :obj:`tarantool.BoxError`

    :rtype: :obj:`dict`

    :raises: :exc:`KeyError`
    """

    stack = []

    while err is not None:
        dict_item = {
            MP_ERROR_TYPE: err.type,
            MP_ERROR_FILE: err.file,
            MP_ERROR_LINE: err.line,
            MP_ERROR_MESSAGE: err.message,
            MP_ERROR_ERRNO: err.errno,
            MP_ERROR_ERRCODE: err.errcode,
        }

        if err.fields is not None:  # omitted if empty
            dict_item[MP_ERROR_FIELDS] = err.fields

        stack.append(dict_item)

        err = err.prev

    return {MP_ERROR_STACK: stack}