Type2Helper.enumerate() - Code Metrics - Inspection of "Release: 0.0.29" - PyETLT/etlt - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9078dd...5d971b )

by P.R.

created 2016-07-23 19:23 UTC

Type2Helper.enumerate() A

↳ Parent: Type2Helper

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	3

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	14
ccs	8
cts	8
cp	1
rs	9.4285
cc	3
crap	3

"""
ETLT

Copyright 2016 Set Based IT Consultancy

Licence MIT
"""
import copy
import datetime


class Type2Helper:
    """
    A helper class for reference data with date intervals.
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, key_start_date, key_end_date, natural_key):
        """
        Object constructor.

        :param str key_start_date: The key of the start date in the rows.
        :param str key_end_date: The key of the end date in the rows.
        :param list[str] natural_key: The keys of the columns that form the natural key.
        """
        self._natural_key = list(natural_key)
        """
        The keys of the columns that form the natural key.

        :type list[str]:
        """

        self._key_end_date = key_end_date
        """
        The key of the end date in the rows.

        :type str:
        """
        self._key_start_date = key_start_date
        """
        The key of the start date in the rows.

        :type str:
        """

        self.rows = dict()
        """
        The data set.

        :type dict:
        """

        self._date_type = ''
        """
        The type of the date fields.
        - date for datetime.date objects
        - str  for strings in ISO 8601 (YYYY-MM-DD) format.

        :type str:
        """

    # ------------------------------------------------------------------------------------------------------------------
    def _get_natural_key(self, row):
        """
        Returns the natural key in a row.

        :param dict row: The row.

        :rtype: tuple
        """
        ret = list()
        for key in self._natural_key:
            ret.append(row[key])

        return tuple(ret)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def _date2int(date):
        """
        Returns an integer representation of a date.

        :param str|datetime.date date: The date.

        :rtype: int
        """
        if isinstance(date, str):
            tmp = datetime.datetime.strptime(date, '%Y-%m-%d')
            return tmp.toordinal()

        if isinstance(date, datetime.date):
            return date.toordinal()

        if isinstance(date, int):
            return date

        raise ValueError('Unexpected type {0!s}'.format(date.__class__))

    # ------------------------------------------------------------------------------------------------------------------
    def _rows_date2int(self, rows):
        """
        Replaces start and end dates in a row set with their integer representation

        :param list[dict[str,T]] rows: The list of rows.

        :rtype: list[dict[str,T]]
        """
        ret = list()
        for row in rows:
            # Make a copy of the row such that self._rows is not affected by merge.
            tmp = copy.copy(row)

            # Determine the type of dates based on the first start date.
            if not self._date_type:
                self._date_type = self._get_date_type(tmp[self._key_start_date])

            # Convert dates to integers.
            tmp[self._key_start_date] = self._date2int(tmp[self._key_start_date])
            tmp[self._key_end_date] = self._date2int(tmp[self._key_end_date])
            ret.append(tmp)

        return ret

    # ------------------------------------------------------------------------------------------------------------------
    def _rows_int2date(self, rows):
        """
        Replaces start and end dates in the row set with their integer representation

        :param list[dict[str,T]] rows: The list of rows.
        """
        for row in rows:
            if self._date_type == 'str':
                row[self._key_start_date] = datetime.date.fromordinal(row[self._key_start_date]).isoformat()
                row[self._key_end_date] = datetime.date.fromordinal(row[self._key_end_date]).isoformat()
            elif self._date_type == 'date':
                row[self._key_start_date] = datetime.date.fromordinal(row[self._key_start_date])
                row[self._key_end_date] = datetime.date.fromordinal(row[self._key_end_date])
            else:
                raise ValueError('Unexpected date type {0!s}'.format(self._date_type))

    # ------------------------------------------------------------------------------------------------------------------
    def _rows_sort(self, rows):
        """
        Returns a list of rows sorted by start and end date.

        :param list[dict[str,T]] rows: The list of rows.

        :rtype: list[dict[str,T]]
        """
        return sorted(rows, key=lambda row: (row[self._key_start_date], row[self._key_end_date]))

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def _get_date_type(date):
        """
        Returns the type of a date.

        :param str|datetime.date date: The date.

        :rtype: str
        """
        if isinstance(date, str):
            return 'str'

        if isinstance(date, datetime.date):
            return 'date'

        raise ValueError('Unexpected type {0!s}'.format(date.__class__))

    # ------------------------------------------------------------------------------------------------------------------
    def enumerate(self, name, start=1):
        """
        Enumerates all rows such that the natural key and the ordinal number are a unique key.

        :param str name: The key holding the ordinal number.
        :param start: The start of the ordinal numbers. Foreach natural key the first row has this ordinal number.
        """
        for natural_key, rows in self.rows.items():
            rows = self._rows_sort(rows)
            ordinal = start
            for row in rows:
                row[name] = ordinal
                ordinal += 1
            self.rows[natural_key] = rows

    # ------------------------------------------------------------------------------------------------------------------
    def prepare_data(self, rows):
        """
        Sets and prepares the rows. The rows are stored in groups in a dictionary. A group is a list of rows with the
        same natural key. The key in the dictionary is a tuple with the values of the natural key.

        :param list[dict] rows: The rows
        """
        self.rows = dict()
        for row in rows:
            natural_key = self._get_natural_key(row)
            if natural_key not in self.rows:
                self.rows[natural_key] = list()
            self.rows[natural_key].append(row)

# ----------------------------------------------------------------------------------------------------------------------


1		"""
2		ETLT
3
4		Copyright 2016 Set Based IT Consultancy
5
6		Licence MIT
7		"""
8	1	import copy
9	1	import datetime
10
11
12	1	class Type2Helper:
13		"""
14		A helper class for reference data with date intervals.
15		"""
16
17		# ------------------------------------------------------------------------------------------------------------------
18	1	def __init__(self, key_start_date, key_end_date, natural_key):
19		"""
20		Object constructor.
21
22		:param str key_start_date: The key of the start date in the rows.
23		:param str key_end_date: The key of the end date in the rows.
24		:param list[str] natural_key: The keys of the columns that form the natural key.
25		"""
26	1	self._natural_key = list(natural_key)
27		"""
28		The keys of the columns that form the natural key.
29
30		:type list[str]:
31		"""
32
33	1	self._key_end_date = key_end_date
34		"""
35		The key of the end date in the rows.
36
37		:type str:
38		"""
39	1	self._key_start_date = key_start_date
40		"""
41		The key of the start date in the rows.
42
43		:type str:
44		"""
45
46	1	self.rows = dict()
47		"""
48		The data set.
49
50		:type dict:
51		"""
52
53	1	self._date_type = ''
54	1	"""
55		The type of the date fields.
56		- date for datetime.date objects
57		- str for strings in ISO 8601 (YYYY-MM-DD) format.
58
59		:type str:
60		"""
61
62		# ------------------------------------------------------------------------------------------------------------------
63	1	def _get_natural_key(self, row):
64		"""
65		Returns the natural key in a row.
66
67		:param dict row: The row.
68
69		:rtype: tuple
70		"""
71	1	ret = list()
72	1	for key in self._natural_key:
73	1	ret.append(row[key])
74
75	1	return tuple(ret)
76
77		# ------------------------------------------------------------------------------------------------------------------
78	1	@staticmethod
79		def _date2int(date):
80		"""
81		Returns an integer representation of a date.
82
83		:param str\|datetime.date date: The date.
84
85		:rtype: int
86		"""
87	1	if isinstance(date, str):
88	1	tmp = datetime.datetime.strptime(date, '%Y-%m-%d')
89	1	return tmp.toordinal()
90
91		if isinstance(date, datetime.date):
92		return date.toordinal()
93
94		if isinstance(date, int):
95		return date
96
97		raise ValueError('Unexpected type {0!s}'.format(date.__class__))
98
99		# ------------------------------------------------------------------------------------------------------------------
100	1	def _rows_date2int(self, rows):
101		"""
102		Replaces start and end dates in a row set with their integer representation
103
104		:param list[dict[str,T]] rows: The list of rows.
105
106		:rtype: list[dict[str,T]]
107		"""
108	1	ret = list()
109	1	for row in rows:
110		# Make a copy of the row such that self._rows is not affected by merge.
111	1	tmp = copy.copy(row)
112
113		# Determine the type of dates based on the first start date.
114	1	if not self._date_type:
115	1	self._date_type = self._get_date_type(tmp[self._key_start_date])
116
117		# Convert dates to integers.
118	1	tmp[self._key_start_date] = self._date2int(tmp[self._key_start_date])
119	1	tmp[self._key_end_date] = self._date2int(tmp[self._key_end_date])
120	1	ret.append(tmp)
121
122	1	return ret
123
124		# ------------------------------------------------------------------------------------------------------------------
125	1	def _rows_int2date(self, rows):
126		"""
127		Replaces start and end dates in the row set with their integer representation
128
129		:param list[dict[str,T]] rows: The list of rows.
130		"""
131	1	for row in rows:
132	1	if self._date_type == 'str':
133	1	row[self._key_start_date] = datetime.date.fromordinal(row[self._key_start_date]).isoformat()
134	1	row[self._key_end_date] = datetime.date.fromordinal(row[self._key_end_date]).isoformat()
135		elif self._date_type == 'date':
136		row[self._key_start_date] = datetime.date.fromordinal(row[self._key_start_date])
137		row[self._key_end_date] = datetime.date.fromordinal(row[self._key_end_date])
138		else:
139		raise ValueError('Unexpected date type {0!s}'.format(self._date_type))
140
141		# ------------------------------------------------------------------------------------------------------------------
142	1	def _rows_sort(self, rows):
143		"""
144		Returns a list of rows sorted by start and end date.
145
146		:param list[dict[str,T]] rows: The list of rows.
147
148		:rtype: list[dict[str,T]]
149		"""
150	1	return sorted(rows, key=lambda row: (row[self._key_start_date], row[self._key_end_date]))
151
152		# ------------------------------------------------------------------------------------------------------------------
153	1	@staticmethod
154		def _get_date_type(date):
155		"""
156		Returns the type of a date.
157
158		:param str\|datetime.date date: The date.
159
160		:rtype: str
161		"""
162	1	if isinstance(date, str):
163	1	return 'str'
164
165		if isinstance(date, datetime.date):
166		return 'date'
167
168		raise ValueError('Unexpected type {0!s}'.format(date.__class__))
169
170		# ------------------------------------------------------------------------------------------------------------------
171	1	def enumerate(self, name, start=1):
172		"""
173		Enumerates all rows such that the natural key and the ordinal number are a unique key.
174
175		:param str name: The key holding the ordinal number.
176		:param start: The start of the ordinal numbers. Foreach natural key the first row has this ordinal number.
177		"""
178	1	for natural_key, rows in self.rows.items():
179	1	rows = self._rows_sort(rows)
180	1	ordinal = start
181	1	for row in rows:
182	1	row[name] = ordinal
183	1	ordinal += 1
184	1	self.rows[natural_key] = rows
185
186		# ------------------------------------------------------------------------------------------------------------------
187	1	def prepare_data(self, rows):
188		"""
189		Sets and prepares the rows. The rows are stored in groups in a dictionary. A group is a list of rows with the
190		same natural key. The key in the dictionary is a tuple with the values of the natural key.
191
192		:param list[dict] rows: The rows
193		"""
194	1	self.rows = dict()
195	1	for row in rows:
196	1	natural_key = self._get_natural_key(row)
197	1	if natural_key not in self.rows:
198	1	self.rows[natural_key] = list()
199	1	self.rows[natural_key].append(row)
200
201		# ----------------------------------------------------------------------------------------------------------------------
202

PyETLT / etlt

Push — master ( 9078dd...5d971b )

Type2Helper.enumerate() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like