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