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
|
# sqlalchemy/processors.py
# Copyright (C) 2010-2021 the SQLAlchemy authors and contributors
# <see AUTHORS file>
# Copyright (C) 2010 Gaetan de Menten gdementen@gmail.com
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""defines generic type conversion functions, as used in bind and result
processors.
They all share one common characteristic: None is passed through unchanged.
"""
import datetime
import re
from . import util
def str_to_datetime_processor_factory(regexp, type_):
rmatch = regexp.match
# Even on python2.6 datetime.strptime is both slower than this code
# and it does not support microseconds.
has_named_groups = bool(regexp.groupindex)
def process(value):
if value is None:
return None
else:
try:
m = rmatch(value)
except TypeError as err:
util.raise_(
ValueError(
"Couldn't parse %s string '%r' "
"- value is not a string." % (type_.__name__, value)
),
from_=err,
)
if m is None:
raise ValueError(
"Couldn't parse %s string: "
"'%s'" % (type_.__name__, value)
)
if has_named_groups:
groups = m.groupdict(0)
return type_(
**dict(
list(
zip(
iter(groups.keys()),
list(map(int, iter(groups.values()))),
)
)
)
)
else:
return type_(*list(map(int, m.groups(0))))
return process
def py_fallback():
def to_decimal_processor_factory(target_class, scale):
fstring = "%%.%df" % scale
def process(value):
if value is None:
return None
else:
return target_class(fstring % value)
return process
def to_float(value): # noqa
if value is None:
return None
else:
return float(value)
def to_str(value): # noqa
if value is None:
return None
else:
return str(value)
def int_to_boolean(value): # noqa
if value is None:
return None
else:
return bool(value)
DATETIME_RE = re.compile(
r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)(?:\.(\d+))?"
)
TIME_RE = re.compile(r"(\d+):(\d+):(\d+)(?:\.(\d+))?")
DATE_RE = re.compile(r"(\d+)-(\d+)-(\d+)")
str_to_datetime = str_to_datetime_processor_factory( # noqa
DATETIME_RE, datetime.datetime
)
str_to_time = str_to_datetime_processor_factory( # noqa
TIME_RE, datetime.time
) # noqa
str_to_date = str_to_datetime_processor_factory( # noqa
DATE_RE, datetime.date
) # noqa
return locals()
try:
from sqlalchemy.cprocessors import DecimalResultProcessor # noqa
from sqlalchemy.cprocessors import int_to_boolean # noqa
from sqlalchemy.cprocessors import str_to_date # noqa
from sqlalchemy.cprocessors import str_to_datetime # noqa
from sqlalchemy.cprocessors import str_to_time # noqa
from sqlalchemy.cprocessors import to_float # noqa
from sqlalchemy.cprocessors import to_str # noqa
def to_decimal_processor_factory(target_class, scale):
# Note that the scale argument is not taken into account for integer
# values in the C implementation while it is in the Python one.
# For example, the Python implementation might return
# Decimal('5.00000') whereas the C implementation will
# return Decimal('5'). These are equivalent of course.
return DecimalResultProcessor(target_class, "%%.%df" % scale).process
except ImportError:
globals().update(py_fallback())
|
