Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pymysql/converters.py: 31%

1import datetime

2import re

3import time

4from decimal import Decimal

6from .constants import FIELD_TYPE

7from .err import ProgrammingError

10def escape_item(val, charset, mapping=None):

11 if mapping is None:

12 mapping = encoders

13 encoder = mapping.get(type(val))

15 # Fallback to default when no encoder found

16 if not encoder:

17 try:

18 encoder = mapping[str]

19 except KeyError:

20 raise TypeError("no default type converter defined")

22 if encoder in (escape_dict, escape_sequence):

23 val = encoder(val, charset, mapping)

24 else:

25 val = encoder(val, mapping)

26 return val

29def escape_dict(val, charset, mapping=None):

30 raise TypeError("dict can not be used as parameter")

33def escape_sequence(val, charset, mapping=None):

34 n = []

35 for item in val:

36 quoted = escape_item(item, charset, mapping)

37 n.append(quoted)

38 return "(" + ",".join(n) + ")"

41def escape_set(val, charset, mapping=None):

42 return ",".join([escape_item(x, charset, mapping) for x in val])

45def escape_bool(value, mapping=None):

46 return str(int(value))

49def escape_int(value, mapping=None):

50 return str(value)

53def escape_float(value, mapping=None):

54 s = repr(value)

55 if s in ("inf", "-inf", "nan"):

56 raise ProgrammingError("%s can not be used with MySQL" % s)

57 if "e" not in s:

58 s += "e0"

59 return s

62_escape_table = [chr(x) for x in range(128)]

63_escape_table[0] = "\\0"

64_escape_table[ord("\\")] = "\\\\"

65_escape_table[ord("\n")] = "\\n"

66_escape_table[ord("\r")] = "\\r"

67_escape_table[ord("\032")] = "\\Z"

68_escape_table[ord('"')] = '\\"'

69_escape_table[ord("'")] = "\\'"

72def escape_string(value, mapping=None):

73 """escapes *value* without adding quote.

75 Value should be unicode

76 """

77 return value.translate(_escape_table)

80def escape_bytes_prefixed(value, mapping=None):

81 return "_binary'%s'" % value.decode("ascii", "surrogateescape").translate(

82 _escape_table

83 )

86def escape_bytes(value, mapping=None):

87 return "'%s'" % value.decode("ascii", "surrogateescape").translate(_escape_table)

90def escape_str(value, mapping=None):

91 return "'%s'" % escape_string(str(value), mapping)

94def escape_None(value, mapping=None):

95 return "NULL"

98def escape_timedelta(obj, mapping=None):

99 # timedelta normalizes a negative value so that seconds/microseconds are

100 # non-negative and the sign lives entirely in days, so the signed magnitude

101 # must be reconstructed before it is split into hours/minutes/seconds --

102 # otherwise a negative TIME comes out with complemented sub-hour fields.

103 sign = ""

104 if obj.days < 0:

105 sign = "-"

106 obj = abs(obj)

107

108 micros = obj.microseconds

109 minutes, seconds = divmod(obj.seconds, 60)

110 hours, minutes = divmod(minutes, 60)

111 hours += obj.days * 24

112

113 if micros:

114 return f"'{sign}{hours:02d}:{minutes:02d}:{seconds:02d}.{micros:06d}'"

115 return f"'{sign}{hours:02d}:{minutes:02d}:{seconds:02d}'"

116

117

118def escape_time(obj, mapping=None):

119 if obj.microsecond:

120 fmt = "'{0.hour:02}:{0.minute:02}:{0.second:02}.{0.microsecond:06}'"

121 else:

122 fmt = "'{0.hour:02}:{0.minute:02}:{0.second:02}'"

123 return fmt.format(obj)

124

125

126def escape_datetime(obj, mapping=None):

127 if obj.microsecond:

128 fmt = (

129 "'{0.year:04}-{0.month:02}-{0.day:02}"

130 + " {0.hour:02}:{0.minute:02}:{0.second:02}.{0.microsecond:06}'"

131 )

132 else:

133 fmt = "'{0.year:04}-{0.month:02}-{0.day:02} {0.hour:02}:{0.minute:02}:{0.second:02}'"

134 return fmt.format(obj)

135

136

137def escape_date(obj, mapping=None):

138 fmt = "'{0.year:04}-{0.month:02}-{0.day:02}'"

139 return fmt.format(obj)

140

141

142def escape_struct_time(obj, mapping=None):

143 return escape_datetime(datetime.datetime(*obj[:6]))

144

145

146def Decimal2Literal(o, d):

147 if not o.is_finite():

148 raise ProgrammingError("%s can not be used with MySQL" % str(o).lower())

149 return format(o, "f")

150

151

152def _convert_second_fraction(s):

153 if not s:

154 return 0

155 # Pad zeros to ensure the fraction length in microseconds

156 s = s.ljust(6, "0")

157 return int(s[:6])

158

159

160DATETIME_RE = re.compile(

161 r"(\d{1,4})-(\d{1,2})-(\d{1,2})[T ](\d{1,2}):(\d{1,2}):(\d{1,2})(?:.(\d{1,6}))?"

162)

163

164

165def convert_datetime(obj):

166 """Returns a DATETIME or TIMESTAMP column value as a datetime object:

167

168 >>> convert_datetime('2007-02-25 23:06:20')

169 datetime.datetime(2007, 2, 25, 23, 6, 20)

170 >>> convert_datetime('2007-02-25T23:06:20')

171 datetime.datetime(2007, 2, 25, 23, 6, 20)

172

173 Illegal values are returned as str:

174

175 >>> convert_datetime('2007-02-31T23:06:20')

176 '2007-02-31T23:06:20'

177 >>> convert_datetime('0000-00-00 00:00:00')

178 '0000-00-00 00:00:00'

179 """

180 if isinstance(obj, (bytes, bytearray)):

181 obj = obj.decode("ascii")

182

183 m = DATETIME_RE.match(obj)

184 if not m:

185 return convert_date(obj)

186

187 try:

188 groups = list(m.groups())

189 groups[-1] = _convert_second_fraction(groups[-1])

190 return datetime.datetime(*[int(x) for x in groups])

191 except ValueError:

192 return convert_date(obj)

193

194

195TIMEDELTA_RE = re.compile(r"(-)?(\d{1,3}):(\d{1,2}):(\d{1,2})(?:.(\d{1,6}))?")

196

197

198def convert_timedelta(obj):

199 """Returns a TIME column as a timedelta object:

200

201 >>> convert_timedelta('25:06:17')

202 datetime.timedelta(days=1, seconds=3977)

203 >>> convert_timedelta('-25:06:17')

204 datetime.timedelta(days=-2, seconds=82423)

205

206 Illegal values are returned as string:

207

208 >>> convert_timedelta('random crap')

209 'random crap'

210

211 Note that MySQL always returns TIME columns as (+|-)HH:MM:SS, but

212 can accept values as (+|-)DD HH:MM:SS. The latter format will not

213 be parsed correctly by this function.

214 """

215 if isinstance(obj, (bytes, bytearray)):

216 obj = obj.decode("ascii")

217

218 m = TIMEDELTA_RE.match(obj)

219 if not m:

220 return obj

221

222 try:

223 groups = list(m.groups())

224 groups[-1] = _convert_second_fraction(groups[-1])

225 negate = -1 if groups[0] else 1

226 hours, minutes, seconds, microseconds = groups[1:]

227

228 tdelta = (

229 datetime.timedelta(

230 hours=int(hours),

231 minutes=int(minutes),

232 seconds=int(seconds),

233 microseconds=int(microseconds),

234 )

235 * negate

236 )

237 return tdelta

238 except ValueError:

239 return obj

240

241

242TIME_RE = re.compile(r"(\d{1,2}):(\d{1,2}):(\d{1,2})(?:.(\d{1,6}))?")

243

244

245def convert_time(obj):

246 """Returns a TIME column as a time object:

247

248 >>> convert_time('15:06:17')

249 datetime.time(15, 6, 17)

250

251 Illegal values are returned as str:

252

253 >>> convert_time('-25:06:17')

254 '-25:06:17'

255 >>> convert_time('random crap')

256 'random crap'

257

258 Note that MySQL always returns TIME columns as (+|-)HH:MM:SS, but

259 can accept values as (+|-)DD HH:MM:SS. The latter format will not

260 be parsed correctly by this function.

261

262 Also note that MySQL's TIME column corresponds more closely to

263 Python's timedelta and not time. However if you want TIME columns

264 to be treated as time-of-day and not a time offset, then you can

265 use set this function as the converter for FIELD_TYPE.TIME.

266 """

267 if isinstance(obj, (bytes, bytearray)):

268 obj = obj.decode("ascii")

269

270 m = TIME_RE.match(obj)

271 if not m:

272 return obj

273

274 try:

275 groups = list(m.groups())

276 groups[-1] = _convert_second_fraction(groups[-1])

277 hours, minutes, seconds, microseconds = groups

278 return datetime.time(

279 hour=int(hours),

280 minute=int(minutes),

281 second=int(seconds),

282 microsecond=int(microseconds),

283 )

284 except ValueError:

285 return obj

286

287

288def convert_date(obj):

289 """Returns a DATE column as a date object:

290

291 >>> convert_date('2007-02-26')

292 datetime.date(2007, 2, 26)

293

294 Illegal values are returned as str:

295

296 >>> convert_date('2007-02-31')

297 '2007-02-31'

298 >>> convert_date('0000-00-00')

299 '0000-00-00'

300 """

301 if isinstance(obj, (bytes, bytearray)):

302 obj = obj.decode("ascii")

303 try:

304 return datetime.date(*[int(x) for x in obj.split("-", 2)])

305 except ValueError:

306 return obj

307

308

309def through(x):

310 return x

311

312

313# def convert_bit(b):

314# b = "\x00" * (8 - len(b)) + b # pad w/ zeroes

315# return struct.unpack(">Q", b)[0]

316#

317# the snippet above is right, but MySQLdb doesn't process bits,

318# so we shouldn't either

319convert_bit = through

320

321

322encoders = {

323 bool: escape_bool,

324 int: escape_int,

325 float: escape_float,

326 str: escape_str,

327 bytes: escape_bytes,

328 tuple: escape_sequence,

329 list: escape_sequence,

330 set: escape_sequence,

331 frozenset: escape_sequence,

332 dict: escape_dict,

333 type(None): escape_None,

334 datetime.date: escape_date,

335 datetime.datetime: escape_datetime,

336 datetime.timedelta: escape_timedelta,

337 datetime.time: escape_time,

338 time.struct_time: escape_struct_time,

339 Decimal: Decimal2Literal,

340}

341

342

343decoders = {

344 FIELD_TYPE.BIT: convert_bit,

345 FIELD_TYPE.TINY: int,

346 FIELD_TYPE.SHORT: int,

347 FIELD_TYPE.LONG: int,

348 FIELD_TYPE.FLOAT: float,

349 FIELD_TYPE.DOUBLE: float,

350 FIELD_TYPE.LONGLONG: int,

351 FIELD_TYPE.INT24: int,

352 FIELD_TYPE.YEAR: int,

353 FIELD_TYPE.TIMESTAMP: convert_datetime,

354 FIELD_TYPE.DATETIME: convert_datetime,

355 FIELD_TYPE.TIME: convert_timedelta,

356 FIELD_TYPE.DATE: convert_date,

357 FIELD_TYPE.BLOB: through,

358 FIELD_TYPE.TINY_BLOB: through,

359 FIELD_TYPE.MEDIUM_BLOB: through,

360 FIELD_TYPE.LONG_BLOB: through,

361 FIELD_TYPE.STRING: through,

362 FIELD_TYPE.VAR_STRING: through,

363 FIELD_TYPE.VARCHAR: through,

364 FIELD_TYPE.DECIMAL: Decimal,

365 FIELD_TYPE.NEWDECIMAL: Decimal,

366}

367

368

369# for MySQLdb compatibility

370conversions = encoders.copy()

371conversions.update(decoders)

372Thing2Literal = escape_str

373

374# Run doctests with `pytest --doctest-modules pymysql/converters.py`