Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/processors.py: 24%

1# sqlalchemy/processors.py 

2# Copyright (C) 2010-2023 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# Copyright (C) 2010 Gaetan de Menten gdementen@gmail.com 

5# 

6# This module is part of SQLAlchemy and is released under 

7# the MIT License: https://www.opensource.org/licenses/mit-license.php 

8 

9"""defines generic type conversion functions, as used in bind and result 

10processors. 

11 

12They all share one common characteristic: None is passed through unchanged. 

13 

14""" 

15 

16import codecs 

17import datetime 

18import re 

19 

20from . import util 

21 

22 

23def str_to_datetime_processor_factory(regexp, type_): 

24 rmatch = regexp.match 

25 # Even on python2.6 datetime.strptime is both slower than this code 

26 # and it does not support microseconds. 

27 has_named_groups = bool(regexp.groupindex) 

28 

29 def process(value): 

30 if value is None: 

31 return None 

32 else: 

33 try: 

34 m = rmatch(value) 

35 except TypeError as err: 

36 util.raise_( 

37 ValueError( 

38 "Couldn't parse %s string '%r' " 

39 "- value is not a string." % (type_.__name__, value) 

40 ), 

41 from_=err, 

42 ) 

43 if m is None: 

44 raise ValueError( 

45 "Couldn't parse %s string: " 

46 "'%s'" % (type_.__name__, value) 

47 ) 

48 if has_named_groups: 

49 groups = m.groupdict(0) 

50 return type_( 

51 **dict( 

52 list( 

53 zip( 

54 iter(groups.keys()), 

55 list(map(int, iter(groups.values()))), 

56 ) 

57 ) 

58 ) 

59 ) 

60 else: 

61 return type_(*list(map(int, m.groups(0)))) 

62 

63 return process 

64 

65 

66def py_fallback(): 

67 def to_unicode_processor_factory(encoding, errors=None): 

68 decoder = codecs.getdecoder(encoding) 

69 

70 def process(value): 

71 if value is None: 

72 return None 

73 else: 

74 # decoder returns a tuple: (value, len). Simply dropping the 

75 # len part is safe: it is done that way in the normal 

76 # 'xx'.decode(encoding) code path. 

77 return decoder(value, errors)[0] 

78 

79 return process 

80 

81 def to_conditional_unicode_processor_factory(encoding, errors=None): 

82 decoder = codecs.getdecoder(encoding) 

83 

84 def process(value): 

85 if value is None: 

86 return None 

87 elif isinstance(value, util.text_type): 

88 return value 

89 else: 

90 # decoder returns a tuple: (value, len). Simply dropping the 

91 # len part is safe: it is done that way in the normal 

92 # 'xx'.decode(encoding) code path. 

93 return decoder(value, errors)[0] 

94 

95 return process 

96 

97 def to_decimal_processor_factory(target_class, scale): 

98 fstring = "%%.%df" % scale 

99 

100 def process(value): 

101 if value is None: 

102 return None 

103 else: 

104 return target_class(fstring % value) 

105 

106 return process 

107 

108 def to_float(value): # noqa 

109 if value is None: 

110 return None 

111 else: 

112 return float(value) 

113 

114 def to_str(value): # noqa 

115 if value is None: 

116 return None 

117 else: 

118 return str(value) 

119 

120 def int_to_boolean(value): # noqa 

121 if value is None: 

122 return None 

123 else: 

124 return bool(value) 

125 

126 DATETIME_RE = re.compile( 

127 r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)(?:\.(\d+))?" 

128 ) 

129 TIME_RE = re.compile(r"(\d+):(\d+):(\d+)(?:\.(\d+))?") 

130 DATE_RE = re.compile(r"(\d+)-(\d+)-(\d+)") 

131 

132 str_to_datetime = str_to_datetime_processor_factory( # noqa 

133 DATETIME_RE, datetime.datetime 

134 ) 

135 str_to_time = str_to_datetime_processor_factory( # noqa 

136 TIME_RE, datetime.time 

137 ) # noqa 

138 str_to_date = str_to_datetime_processor_factory( # noqa 

139 DATE_RE, datetime.date 

140 ) # noqa 

141 return locals() 

142 

143 

144try: 

145 from sqlalchemy.cprocessors import DecimalResultProcessor # noqa 

146 from sqlalchemy.cprocessors import int_to_boolean # noqa 

147 from sqlalchemy.cprocessors import str_to_date # noqa 

148 from sqlalchemy.cprocessors import str_to_datetime # noqa 

149 from sqlalchemy.cprocessors import str_to_time # noqa 

150 from sqlalchemy.cprocessors import to_float # noqa 

151 from sqlalchemy.cprocessors import to_str # noqa 

152 from sqlalchemy.cprocessors import UnicodeResultProcessor # noqa 

153 

154 def to_unicode_processor_factory(encoding, errors=None): 

155 if errors is not None: 

156 return UnicodeResultProcessor(encoding, errors).process 

157 else: 

158 return UnicodeResultProcessor(encoding).process 

159 

160 def to_conditional_unicode_processor_factory(encoding, errors=None): 

161 if errors is not None: 

162 return UnicodeResultProcessor(encoding, errors).conditional_process 

163 else: 

164 return UnicodeResultProcessor(encoding).conditional_process 

165 

166 def to_decimal_processor_factory(target_class, scale): 

167 # Note that the scale argument is not taken into account for integer 

168 # values in the C implementation while it is in the Python one. 

169 # For example, the Python implementation might return 

170 # Decimal('5.00000') whereas the C implementation will 

171 # return Decimal('5'). These are equivalent of course. 

172 return DecimalResultProcessor(target_class, "%%.%df" % scale).process 

173 

174 

175except ImportError: 

176 globals().update(py_fallback())