src.record.Record._parse() - Code Metrics - mattboyer/sqbrite - Measure and Improve Code Quality continuously with Scrutinizer

src.record.Record._parse() C
last analyzed 2022-03-20 10:13 UTC

↳ Parent: src.record

Complexity

Conditions

Size

Total Lines	70
Code Lines	49

Duplication

Lines	70
Ratio	100 %

Importance

Changes

Metric	Value
cc	11
eloc	49
nop	1
dl	70
loc	70
rs	5.4
c	0
b	0
f	0

How to fix Long Method Complexity

# MIT License
#
# Copyright (c) 2017 Matt Boyer
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from . import _LOGGER
from .field import (Field, MalformedField)
from .utils import (Varint, IndexDict)


class MalformedRecord(Exception):
    pass


class Record(object):


    column_types = {
        0: (0, "NULL"),
        1: (1, "8-bit twos-complement integer"),
        2: (2, "big-endian 16-bit twos-complement integer"),
        3: (3, "big-endian 24-bit twos-complement integer"),
        4: (4, "big-endian 32-bit twos-complement integer"),
        5: (6, "big-endian 48-bit twos-complement integer"),
        6: (8, "big-endian 64-bit twos-complement integer"),
        7: (8, "Floating point"),
        8: (0, "Integer 0"),
        9: (0, "Integer 1"),
    }

    def __init__(self, record_bytes):
        self._bytes = record_bytes
        self._header_bytes = None
        self._fields = IndexDict()
        self._parse()

    def __bytes__(self):
        return self._bytes

    @property
    def header(self):
        return self._header_bytes

    @property
    def fields(self):
        return self._fields

    def truncate(self, new_length):
        self._bytes = self._bytes[:new_length]
        self._parse()

    def _parse(self):
        header_offset = 0

        header_length_varint = Varint(
            # A varint is encoded on *at most* 9 bytes
            bytes(self)[header_offset:9 + header_offset]
        )

        # Let's keep track of how many bytes of the Record header (including
        # the header length itself) we've succesfully parsed
        parsed_header_bytes = len(header_length_varint)

        if len(bytes(self)) < int(header_length_varint):
            raise MalformedRecord(
                "Not enough bytes to fully read the record header!"
            )

        header_offset += len(header_length_varint)
        self._header_bytes = bytes(self)[:int(header_length_varint)]

        col_idx = 0
        field_offset = int(header_length_varint)
        while header_offset < int(header_length_varint):
            serial_type_varint = Varint(
                bytes(self)[header_offset:9 + header_offset]
            )
            serial_type = int(serial_type_varint)
            col_length = None

            try:
                col_length, _ = self.column_types[serial_type]
            except KeyError as col_type_ex:
                if serial_type >= 13 and (1 == serial_type % 2):
                    col_length = (serial_type - 13) // 2
                elif serial_type >= 12 and (0 == serial_type % 2):
                    col_length = (serial_type - 12) // 2
                else:
                    raise ValueError(
                        "Unknown serial type {}".format(serial_type)
                    ) from col_type_ex

            try:
                field_obj = Field(
                    col_idx,
                    serial_type,
                    bytes(self)[field_offset:field_offset + col_length]
                )
            except MalformedField as ex:
                _LOGGER.warning(
                    "Caught %r while instantiating field %d (%d)",
                    ex, col_idx, serial_type
                )
                raise MalformedRecord from ex
            except Exception as ex:
                _LOGGER.warning(
                    "Caught %r while instantiating field %d (%d)",
                    ex, col_idx, serial_type
                )
                # pdb.set_trace()
                raise

            self._fields[col_idx] = field_obj
            col_idx += 1
            field_offset += col_length

            parsed_header_bytes += len(serial_type_varint)
            header_offset += len(serial_type_varint)

            if field_offset > len(bytes(self)):
                raise MalformedRecord

        # assert(parsed_header_bytes == int(header_length_varint))

    def print_fields(self, table=None):
        for field_idx in self._fields:
            field_obj = self._fields[field_idx]
            if not table or table.columns is None:
                _LOGGER.info(
                    "\tField %d (%d bytes), type %d: %s",
                    field_obj.index,
                    len(field_obj),
                    field_obj.serial_type,
                    field_obj.value
                )
            else:
                _LOGGER.info(
                    "\t%s: %s",
                    table.columns[field_obj.index],
                    field_obj.value
                )

    def __repr__(self):
        return '<Record {} fields, {} bytes, header: {} bytes>'.format(
            len(self._fields), len(bytes(self)), len(self.header)
        )


1		# MIT License
2		#
3		# Copyright (c) 2017 Matt Boyer
4		#
5		# Permission is hereby granted, free of charge, to any person obtaining a copy
6		# of this software and associated documentation files (the "Software"), to deal
7		# in the Software without restriction, including without limitation the rights
8		# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9		# copies of the Software, and to permit persons to whom the Software is
10		# furnished to do so, subject to the following conditions:
11		#
12		# The above copyright notice and this permission notice shall be included in
13		# all copies or substantial portions of the Software.
14		#
15		# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16		# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17		# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18		# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19		# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20		# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21		# SOFTWARE.
22
23		from . import _LOGGER
24		from .field import (Field, MalformedField)
25		from .utils import (Varint, IndexDict)
26
27
28		class MalformedRecord(Exception):
29		pass
30
31
32	View Code Duplication	class Record(object):
		0 ignored issues – show Duplication introduced 2022-03-20 09:42 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
33
34		column_types = {
35		0: (0, "NULL"),
36		1: (1, "8-bit twos-complement integer"),
37		2: (2, "big-endian 16-bit twos-complement integer"),
38		3: (3, "big-endian 24-bit twos-complement integer"),
39		4: (4, "big-endian 32-bit twos-complement integer"),
40		5: (6, "big-endian 48-bit twos-complement integer"),
41		6: (8, "big-endian 64-bit twos-complement integer"),
42		7: (8, "Floating point"),
43		8: (0, "Integer 0"),
44		9: (0, "Integer 1"),
45		}
46
47		def __init__(self, record_bytes):
48		self._bytes = record_bytes
49		self._header_bytes = None
50		self._fields = IndexDict()
51		self._parse()
52
53		def __bytes__(self):
54		return self._bytes
55
56		@property
57		def header(self):
58		return self._header_bytes
59
60		@property
61		def fields(self):
62		return self._fields
63
64		def truncate(self, new_length):
65		self._bytes = self._bytes[:new_length]
66		self._parse()
67
68		def _parse(self):
69		header_offset = 0
70
71		header_length_varint = Varint(
72		# A varint is encoded on at most 9 bytes
73		bytes(self)[header_offset:9 + header_offset]
74		)
75
76		# Let's keep track of how many bytes of the Record header (including
77		# the header length itself) we've succesfully parsed
78		parsed_header_bytes = len(header_length_varint)
79
80		if len(bytes(self)) < int(header_length_varint):
81		raise MalformedRecord(
82		"Not enough bytes to fully read the record header!"
83		)
84
85		header_offset += len(header_length_varint)
86		self._header_bytes = bytes(self)[:int(header_length_varint)]
87
88		col_idx = 0
89		field_offset = int(header_length_varint)
90		while header_offset < int(header_length_varint):
91		serial_type_varint = Varint(
92		bytes(self)[header_offset:9 + header_offset]
93		)
94		serial_type = int(serial_type_varint)
95		col_length = None
96
97		try:
98		col_length, _ = self.column_types[serial_type]
99		except KeyError as col_type_ex:
100		if serial_type >= 13 and (1 == serial_type % 2):
101		col_length = (serial_type - 13) // 2
102		elif serial_type >= 12 and (0 == serial_type % 2):
103		col_length = (serial_type - 12) // 2
104		else:
105		raise ValueError(
106		"Unknown serial type {}".format(serial_type)
107		) from col_type_ex
108
109		try:
110		field_obj = Field(
111		col_idx,
112		serial_type,
113		bytes(self)[field_offset:field_offset + col_length]
114		)
115		except MalformedField as ex:
116		_LOGGER.warning(
117		"Caught %r while instantiating field %d (%d)",
118		ex, col_idx, serial_type
119		)
120		raise MalformedRecord from ex
121		except Exception as ex:
122		_LOGGER.warning(
123		"Caught %r while instantiating field %d (%d)",
124		ex, col_idx, serial_type
125		)
126		# pdb.set_trace()
127		raise
128
129		self._fields[col_idx] = field_obj
130		col_idx += 1
131		field_offset += col_length
132
133		parsed_header_bytes += len(serial_type_varint)
134		header_offset += len(serial_type_varint)
135
136		if field_offset > len(bytes(self)):
137		raise MalformedRecord
138
139		# assert(parsed_header_bytes == int(header_length_varint))
140
141		def print_fields(self, table=None):
142		for field_idx in self._fields:
143		field_obj = self._fields[field_idx]
144		if not table or table.columns is None:
145		_LOGGER.info(
146		"\tField %d (%d bytes), type %d: %s",
147		field_obj.index,
148		len(field_obj),
149		field_obj.serial_type,
150		field_obj.value
151		)
152		else:
153		_LOGGER.info(
154		"\t%s: %s",
155		table.columns[field_obj.index],
156		field_obj.value
157		)
158
159		def __repr__(self):
160		return '<Record {} fields, {} bytes, header: {} bytes>'.format(
161		len(self._fields), len(bytes(self)), len(self.header)
162		)
163

mattboyer / sqbrite

src.record.Record._parse() C last analyzed 2022-03-20 10:13 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

src.record.Record._parse() C
last analyzed 2022-03-20 10:13 UTC