etlt_mysql.writer.MySqlLoaderWriter.MySqlLoaderWriter.writerow() - Code Metrics - Inspection of "Merge remote-tracking branch 'origin/master'" - PyETLT/etlt-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 715ae9...b2e443 )

by P.R.

created 2020-10-26 17:32 UTC

MySqlLoaderWriter.writerow() A

↳ Parent: etlt_mysql.writer.MySqlLoaderWriter

Complexity

Conditions

Size

Total Lines	11
Code Lines	5

Duplication

Lines	11
Ratio	100 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

Metric	Value
eloc	5
dl	11
loc	11
rs	10
c	0
b	0
f	0
ccs	5
cts	5
cp	1
cc	2
nop	2
crap	2

import datetime
import os
from abc import ABC
from decimal import Decimal
from typing import Any, Dict

import pytz
from etlt.writer.SqlLoaderWriter import SqlLoaderWriter


class MySqlLoaderWriter(SqlLoaderWriter, ABC):

    """
    Writer for storing rows in CSV format optimized for MariaDB and MySQL instances and 'load data local infile'
    statement.
    """

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_bool(value: bool, file: Any) -> None:
        """
        Writes a boolean as a field to a CSV file.

        :param bool value: The boolean.
        :param Any file: The file like object
        """
        if value:
            file.write('1')
        else:
            file.write('0')

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_date(value: datetime.date, file: Any) -> None:
        """
        Writes a date as a field to a CSV file.

        :param datetime.date value: The date.
        :param Any file: The file like object
        """
        file.write(value.isoformat())

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_datetime(value: datetime.datetime, file: Any) -> None:
        """
        Writes a datetime as a field to a CSV file.

        :param datetime.datetime value: The date.
        :param Any file: The file like object
        """
        zone = value.tzinfo
        if zone and zone != pytz.utc:
            raise ValueError('Only native and UTC timezone supported')
        file.write(value.isoformat())

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_timedelta(value: datetime.timedelta, file: Any) -> None:
        """
        Writes a timedelta as a field to a CSV file.

        :param datetime.timedelta value: The timedelta.
        :param Any file: The file like object
        """
        MySqlLoaderWriter.write_string(str(value), file)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_decimal(value: Decimal, file: Any) -> None:
        """
        Writes a decimal as a field to a CSV file.

        :param Decimal value: The decimal.
        :param Any file: The file like object
        """
        file.write(str(value))

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_float(value: float, file: Any) -> None:
        """
        Writes a float as a field to a CSV file.

        :param float value: The float.
        :param Any file: The file like object
        """
        file.write(str(value))

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_int(value: int, file: Any) -> None:
        """
        Writes an integer as a field to a CSV file.

        :param int value: The integer.
        :param Any file: The file like object
        """
        file.write(str(value))

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_none(_, file: Any) -> None:
        """
        Writes None as a field to a CSV file.

        :param None _: The None object.
        :param Any file: The file like object
        """
        file.write('\\N')

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_string(value: str, file: Any) -> None:
        """
        Writes a string as a field to a CSV file.

        :param str value: The string.
        :param Any file: The file.
        """
        if value == '':
            file.write('\\N')
        else:
            file.write("'")
            file.write(value.replace('\\', '\\\\').replace("'", "\\'"))
            file.write("'")

    # ------------------------------------------------------------------------------------------------------------------
    def writerow(self, row: Dict[str, Any]) -> None:
        """
        Writes a row to the destination file.

        :param dict[str,Any] row: The row.
        """
        for field in self._fields:
            self._write_field(row[field])
            self._file.write(',')

        self._file.write(os.linesep)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def write_uuid(value, file: Any) -> None:
        """
        Writes a UUID as a field to a CSV file.

        :param uuid.UUID value: The UUID.
        :param Any file: The file like object
        """
        MySqlLoaderWriter.write_string(str(value), file)

    # ------------------------------------------------------------------------------------------------------------------
    def get_bulk_load_sql(self, table_name: str) -> str:
        """
        Returns a SQL statement for bulk loading the data into a table.

        :param str table_name: The name of the table.

        :rtype: str
        """
        sql = "load data local infile '{FILENAME}'"
        sql += ' into table `{TABLE_NAME}`'
        sql += ' character set {ENCODING}'
        sql += " fields terminated by ','"
        sql += " optionally enclosed by '\\\''"
        sql += " escaped by '\\\\'"
        sql += " lines terminated by '\\n'"

        return sql.format(FILENAME=self._filename,  # @todo Quoting of filename
                          ENCODING=self._encoding,
                          TABLE_NAME=table_name)


# ----------------------------------------------------------------------------------------------------------------------
MySqlLoaderWriter.register_handler("<class 'bool'>", MySqlLoaderWriter.write_bool)
MySqlLoaderWriter.register_handler("<class 'datetime.date'>", MySqlLoaderWriter.write_date)
MySqlLoaderWriter.register_handler("<class 'datetime.datetime'>", MySqlLoaderWriter.write_datetime)
MySqlLoaderWriter.register_handler("<class 'datetime.timedelta'>", MySqlLoaderWriter.write_timedelta)
MySqlLoaderWriter.register_handler("<class 'decimal.Decimal'>", MySqlLoaderWriter.write_decimal)
MySqlLoaderWriter.register_handler("<class 'float'>", MySqlLoaderWriter.write_float)
MySqlLoaderWriter.register_handler("<class 'int'>", MySqlLoaderWriter.write_int)
MySqlLoaderWriter.register_handler("<class 'NoneType'>", MySqlLoaderWriter.write_none)
MySqlLoaderWriter.register_handler("<class 'str'>", MySqlLoaderWriter.write_string)
MySqlLoaderWriter.register_handler("<class 'uuid.UUID'>", MySqlLoaderWriter.write_uuid)


1	1		import datetime
2	1		import os
3	1		from abc import ABC
4	1		from decimal import Decimal
5	1		from typing import Any, Dict
6
7	1		import pytz
8	1		from etlt.writer.SqlLoaderWriter import SqlLoaderWriter
9
10
11	1	View Code Duplication	class MySqlLoaderWriter(SqlLoaderWriter, ABC):
			0 ignored issues – show Duplication introduced 2020-10-26 17:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
12			"""
13			Writer for storing rows in CSV format optimized for MariaDB and MySQL instances and 'load data local infile'
14			statement.
15			"""
16
17			# ------------------------------------------------------------------------------------------------------------------
18	1		@staticmethod
19	1		def write_bool(value: bool, file: Any) -> None:
20			"""
21			Writes a boolean as a field to a CSV file.
22
23			:param bool value: The boolean.
24			:param Any file: The file like object
25			"""
26	1		if value:
27	1		file.write('1')
28			else:
29	1		file.write('0')
30
31			# ------------------------------------------------------------------------------------------------------------------
32	1		@staticmethod
33	1		def write_date(value: datetime.date, file: Any) -> None:
34			"""
35			Writes a date as a field to a CSV file.
36
37			:param datetime.date value: The date.
38			:param Any file: The file like object
39			"""
40	1		file.write(value.isoformat())
41
42			# ------------------------------------------------------------------------------------------------------------------
43	1		@staticmethod
44	1		def write_datetime(value: datetime.datetime, file: Any) -> None:
45			"""
46			Writes a datetime as a field to a CSV file.
47
48			:param datetime.datetime value: The date.
49			:param Any file: The file like object
50			"""
51	1		zone = value.tzinfo
52	1		if zone and zone != pytz.utc:
53			raise ValueError('Only native and UTC timezone supported')
54	1		file.write(value.isoformat())
55
56			# ------------------------------------------------------------------------------------------------------------------
57	1		@staticmethod
58	1		def write_timedelta(value: datetime.timedelta, file: Any) -> None:
59			"""
60			Writes a timedelta as a field to a CSV file.
61
62			:param datetime.timedelta value: The timedelta.
63			:param Any file: The file like object
64			"""
65	1		MySqlLoaderWriter.write_string(str(value), file)
66
67			# ------------------------------------------------------------------------------------------------------------------
68	1		@staticmethod
69	1		def write_decimal(value: Decimal, file: Any) -> None:
70			"""
71			Writes a decimal as a field to a CSV file.
72
73			:param Decimal value: The decimal.
74			:param Any file: The file like object
75			"""
76	1		file.write(str(value))
77
78			# ------------------------------------------------------------------------------------------------------------------
79	1		@staticmethod
80	1		def write_float(value: float, file: Any) -> None:
81			"""
82			Writes a float as a field to a CSV file.
83
84			:param float value: The float.
85			:param Any file: The file like object
86			"""
87	1		file.write(str(value))
88
89			# ------------------------------------------------------------------------------------------------------------------
90	1		@staticmethod
91	1		def write_int(value: int, file: Any) -> None:
92			"""
93			Writes an integer as a field to a CSV file.
94
95			:param int value: The integer.
96			:param Any file: The file like object
97			"""
98	1		file.write(str(value))
99
100			# ------------------------------------------------------------------------------------------------------------------
101	1		@staticmethod
102	1		def write_none(_, file: Any) -> None:
103			"""
104			Writes None as a field to a CSV file.
105
106			:param None _: The None object.
107			:param Any file: The file like object
108			"""
109	1		file.write('\\N')
110
111			# ------------------------------------------------------------------------------------------------------------------
112	1		@staticmethod
113	1		def write_string(value: str, file: Any) -> None:
114			"""
115			Writes a string as a field to a CSV file.
116
117			:param str value: The string.
118			:param Any file: The file.
119			"""
120	1		if value == '':
121	1		file.write('\\N')
122			else:
123	1		file.write("'")
124	1		file.write(value.replace('\\', '\\\\').replace("'", "\\'"))
125	1		file.write("'")
126
127			# ------------------------------------------------------------------------------------------------------------------
128	1		def writerow(self, row: Dict[str, Any]) -> None:
129			"""
130			Writes a row to the destination file.
131
132			:param dict[str,Any] row: The row.
133			"""
134	1		for field in self._fields:
135	1		self._write_field(row[field])
136	1		self._file.write(',')
137
138	1		self._file.write(os.linesep)
139
140			# ------------------------------------------------------------------------------------------------------------------
141	1		@staticmethod
142	1		def write_uuid(value, file: Any) -> None:
143			"""
144			Writes a UUID as a field to a CSV file.
145
146			:param uuid.UUID value: The UUID.
147			:param Any file: The file like object
148			"""
149	1		MySqlLoaderWriter.write_string(str(value), file)
150
151			# ------------------------------------------------------------------------------------------------------------------
152	1		def get_bulk_load_sql(self, table_name: str) -> str:
153			"""
154			Returns a SQL statement for bulk loading the data into a table.
155
156			:param str table_name: The name of the table.
157
158			:rtype: str
159			"""
160			sql = "load data local infile '{FILENAME}'"
161			sql += ' into table `{TABLE_NAME}`'
162			sql += ' character set {ENCODING}'
163			sql += " fields terminated by ','"
164			sql += " optionally enclosed by '\\\''"
165			sql += " escaped by '\\\\'"
166			sql += " lines terminated by '\\n'"
167
168			return sql.format(FILENAME=self._filename, # @todo Quoting of filename
169			ENCODING=self._encoding,
170			TABLE_NAME=table_name)
171
172
173			# ----------------------------------------------------------------------------------------------------------------------
174	1		MySqlLoaderWriter.register_handler("<class 'bool'>", MySqlLoaderWriter.write_bool)
175	1		MySqlLoaderWriter.register_handler("<class 'datetime.date'>", MySqlLoaderWriter.write_date)
176	1		MySqlLoaderWriter.register_handler("<class 'datetime.datetime'>", MySqlLoaderWriter.write_datetime)
177	1		MySqlLoaderWriter.register_handler("<class 'datetime.timedelta'>", MySqlLoaderWriter.write_timedelta)
178	1		MySqlLoaderWriter.register_handler("<class 'decimal.Decimal'>", MySqlLoaderWriter.write_decimal)
179	1		MySqlLoaderWriter.register_handler("<class 'float'>", MySqlLoaderWriter.write_float)
180	1		MySqlLoaderWriter.register_handler("<class 'int'>", MySqlLoaderWriter.write_int)
181	1		MySqlLoaderWriter.register_handler("<class 'NoneType'>", MySqlLoaderWriter.write_none)
182	1		MySqlLoaderWriter.register_handler("<class 'str'>", MySqlLoaderWriter.write_string)
183			MySqlLoaderWriter.register_handler("<class 'uuid.UUID'>", MySqlLoaderWriter.write_uuid)
184

PyETLT / etlt-mysql

Push — master ( 715ae9...b2e443 )

MySqlLoaderWriter.writerow() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like