FilesExpand file tree

 master

/

mysql.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

159 lines (131 loc) · 4.55 KB

 master

/

mysql.py

Copy path

Top

File metadata and controls

• Code
• Blame

159 lines (131 loc) · 4.55 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

146

147

148

149

150

151

152

153

154

155

156

157

158

159

from typing import Any, ClassVar, Dict, Type, Union

import attrs

from data_diff.abcs.database_types import (

Datetime,

Timestamp,

Float,

Decimal,

Integer,

Text,

TemporalType,

FractionalType,

ColType_UUID,

Boolean,

Date,

)

from data_diff.databases.base import (

ThreadedDatabase,

import_helper,

ConnectError,

BaseDialect,

ThreadLocalInterpreter,

)

from data_diff.databases.base import (

MD5_HEXDIGITS,

CHECKSUM_HEXDIGITS,

TIMESTAMP_PRECISION_POS,

CHECKSUM_OFFSET,

)

@import_helper("mysql")

def import_mysql():

import mysql.connector

return mysql.connector

@attrs.define(frozen=False)

class Dialect(BaseDialect):

name = "MySQL"

ROUNDS_ON_PREC_LOSS = True

SUPPORTS_PRIMARY_KEY: ClassVar[bool] = True

SUPPORTS_INDEXES = True

TYPE_CLASSES = {

# Dates

"datetime": Datetime,

"timestamp": Timestamp,

"date": Date,

# Numbers

"double": Float,

"float": Float,

"decimal": Decimal,

"int": Integer,

"bigint": Integer,

"mediumint": Integer,

"smallint": Integer,

"tinyint": Integer,

# Text

"varchar": Text,

"char": Text,

"varbinary": Text,

"binary": Text,

"text": Text,

"mediumtext": Text,

"longtext": Text,

"tinytext": Text,

# Boolean

"boolean": Boolean,

}

def quote(self, s: str) -> str:

return f"`{s}`"

def to_string(self, s: str) -> str:

return f"cast({s} as char)"

def is_distinct_from(self, a: str, b: str) -> str:

return f"not ({a} <=> {b})"

def random(self) -> str:

return "RAND()"

def type_repr(self, t) -> str:

try:

return {

str: "VARCHAR(1024)",

}[t]

except KeyError:

return super().type_repr(t)

def explain_as_text(self, query: str) -> str:

return f"EXPLAIN FORMAT=TREE {query}"

def optimizer_hints(self, s: str):

return f"/*+ {s} */ "

def set_timezone_to_utc(self) -> str:

return "SET @@session.time_zone='+00:00'"

def md5_as_int(self, s: str) -> str:

return f"conv(substring(md5({s}), {1+MD5_HEXDIGITS-CHECKSUM_HEXDIGITS}), 16, 10) - {CHECKSUM_OFFSET}"

def md5_as_hex(self, s: str) -> str:

return f"md5({s})"

def normalize_timestamp(self, value: str, coltype: TemporalType) -> str:

if coltype.rounds:

return self.to_string(f"cast( cast({value} as datetime({coltype.precision})) as datetime(6))")

s = self.to_string(f"cast({value} as datetime(6))")

return f"RPAD(RPAD({s}, {TIMESTAMP_PRECISION_POS+coltype.precision}, '.'), {TIMESTAMP_PRECISION_POS+6}, '0')"

def normalize_number(self, value: str, coltype: FractionalType) -> str:

return self.to_string(f"cast({value} as decimal(38, {coltype.precision}))")

def normalize_uuid(self, value: str, coltype: ColType_UUID) -> str:

return f"TRIM(CAST({value} AS char))"

@attrs.define(frozen=False, init=False, kw_only=True)

class MySQL(ThreadedDatabase):

DIALECT_CLASS: ClassVar[Type[BaseDialect]] = Dialect

SUPPORTS_ALPHANUMS = False

SUPPORTS_UNIQUE_CONSTAINT = True

CONNECT_URI_HELP = "mysql://<user>:<password>@<host>/<database>"

CONNECT_URI_PARAMS = ["database?"]

_args: Dict[str, Any]

def __init__(self, *, thread_count, **kw) -> None:

super().__init__(thread_count=thread_count)

self._args = kw

# In MySQL schema and database are synonymous

try:

self.default_schema = kw["database"]

except KeyError:

raise ValueError("MySQL URL must specify a database")

def create_connection(self):

mysql = import_mysql()

try:

return mysql.connect(charset="utf8", use_unicode=True, **self._args)

except mysql.Error as e:

if e.errno == mysql.errorcode.ER_ACCESS_DENIED_ERROR:

raise ConnectError("Bad user name or password") from e

elif e.errno == mysql.errorcode.ER_BAD_DB_ERROR:

raise ConnectError("Database does not exist") from e

raise ConnectError(*e.args) from e

def _query_in_worker(self, sql_code: Union[str, ThreadLocalInterpreter]):

"This method runs in a worker thread"

if self._init_error:

raise self._init_error

if not self.thread_local.conn.is_connected():

self.thread_local.conn.ping(reconnect=True, attempts=3, delay=5)

return self._query_conn(self.thread_local.conn, sql_code)