Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/processors.py: 51%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

35 statements  

1# engine/processors.py 

2# Copyright (C) 2010-2026 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 

16from __future__ import annotations 

17 

18import datetime 

19from typing import Callable 

20from typing import Optional 

21from typing import Pattern 

22from typing import TypeVar 

23from typing import Union 

24 

25from ._processors_cy import int_to_boolean as int_to_boolean # noqa: F401 

26from ._processors_cy import str_to_date as str_to_date # noqa: F401 

27from ._processors_cy import str_to_datetime as str_to_datetime # noqa: F401 

28from ._processors_cy import str_to_time as str_to_time # noqa: F401 

29from ._processors_cy import to_float as to_float # noqa: F401 

30from ._processors_cy import to_str as to_str # noqa: F401 

31 

32if True: 

33 from ._processors_cy import ( # noqa: F401 

34 to_decimal_processor_factory as to_decimal_processor_factory, 

35 ) 

36 

37 

38_DT = TypeVar( 

39 "_DT", bound=Union[datetime.datetime, datetime.time, datetime.date] 

40) 

41 

42 

43def str_to_datetime_processor_factory( 

44 regexp: Pattern[str], type_: Callable[..., _DT] 

45) -> Callable[[Optional[str]], Optional[_DT]]: 

46 rmatch = regexp.match 

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

48 # and it does not support microseconds. 

49 has_named_groups = bool(regexp.groupindex) 

50 

51 def process(value: Optional[str]) -> Optional[_DT]: 

52 if value is None: 

53 return None 

54 else: 

55 try: 

56 m = rmatch(value) 

57 except TypeError as err: 

58 raise ValueError( 

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

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

61 ) from err 

62 

63 if m is None: 

64 raise ValueError( 

65 "Couldn't parse %s string: " 

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

67 ) 

68 if has_named_groups: 

69 groups = m.groupdict(0) 

70 return type_( 

71 **dict( 

72 list( 

73 zip( 

74 iter(groups.keys()), 

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

76 ) 

77 ) 

78 ) 

79 ) 

80 else: 

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

82 

83 return process