1import decimal
2from collections.abc import Callable, Sequence
3from functools import cached_property
4from typing import TYPE_CHECKING, Any
5
6from plain.preflight import PreflightResult
7from psycopg.types import numeric
8
9from plain import exceptions, validators
10
11from .base import NOT_PROVIDED, DefaultableField
12
13if TYPE_CHECKING:
14 from plain.postgres.connection import DatabaseConnection
15
16
17class FloatField[T: (float, float | None) = float](DefaultableField[T]):
18 db_type_sql = "double precision"
19 empty_strings_allowed = False
20
21 def get_prep_value(self, value: Any) -> Any:
22 value = super().get_prep_value(value)
23 if value is None:
24 return None
25 try:
26 return float(value)
27 except (TypeError, ValueError) as e:
28 raise e.__class__(
29 f"Field '{self.name}' expected a number but got {value!r}.",
30 ) from e
31
32 def to_python(self, value: Any) -> float | None:
33 if value is None:
34 return value
35 try:
36 return float(value)
37 except TypeError, ValueError:
38 raise exceptions.ValidationError(
39 '"%(value)s" value must be a float.',
40 code="invalid",
41 params={"value": value},
42 )
43
44
45class IntegerField[T: (int, int | None) = int](DefaultableField[T]):
46 db_type_sql = "integer"
47 integer_range: tuple[int, int] = (-2147483648, 2147483647)
48 psycopg_type: type = numeric.Int4
49 empty_strings_allowed = False
50
51 @cached_property
52 def validators(self) -> list[Callable[..., Any]]:
53 # These validators can't be added at field initialization time since
54 # they're based on values retrieved from the database connection.
55 validators_ = super().validators
56 min_value, max_value = self.integer_range
57 if min_value is not None and not any(
58 (
59 isinstance(validator, validators.MinValueValidator)
60 and (
61 validator.limit_value()
62 if callable(validator.limit_value)
63 else validator.limit_value
64 )
65 >= min_value
66 )
67 for validator in validators_
68 ):
69 validators_.append(validators.MinValueValidator(min_value))
70 if max_value is not None and not any(
71 (
72 isinstance(validator, validators.MaxValueValidator)
73 and (
74 validator.limit_value()
75 if callable(validator.limit_value)
76 else validator.limit_value
77 )
78 <= max_value
79 )
80 for validator in validators_
81 ):
82 validators_.append(validators.MaxValueValidator(max_value))
83 return validators_
84
85 def get_prep_value(self, value: Any) -> Any:
86 value = super().get_prep_value(value)
87 if value is None:
88 return None
89 try:
90 return int(value)
91 except (TypeError, ValueError) as e:
92 raise e.__class__(
93 f"Field '{self.name}' expected a number but got {value!r}.",
94 ) from e
95
96 def get_db_prep_value(
97 self, value: Any, connection: DatabaseConnection, prepared: bool = False
98 ) -> Any:
99 from plain.postgres.expressions import ResolvableExpression
100
101 value = super().get_db_prep_value(value, connection, prepared)
102 if value is None or isinstance(value, ResolvableExpression):
103 return value
104 return self.psycopg_type(value)
105
106 def to_python(self, value: Any) -> int | None:
107 if value is None:
108 return value
109 try:
110 return int(value)
111 except TypeError, ValueError:
112 raise exceptions.ValidationError(
113 '"%(value)s" value must be an integer.',
114 code="invalid",
115 params={"value": value},
116 )
117
118
119class BigIntegerField[T: (int, int | None) = int](IntegerField[T]):
120 db_type_sql = "bigint"
121 integer_range = (-9223372036854775808, 9223372036854775807)
122 psycopg_type = numeric.Int8
123
124
125class SmallIntegerField[T: (int, int | None) = int](IntegerField[T]):
126 db_type_sql = "smallint"
127 integer_range = (-32768, 32767)
128 psycopg_type = numeric.Int2
129
130
131class DecimalField[T: (decimal.Decimal, decimal.Decimal | None) = decimal.Decimal](
132 DefaultableField[T]
133):
134 db_type_sql = "numeric(%(max_digits)s,%(decimal_places)s)"
135 empty_strings_allowed = False
136
137 def __init__(
138 self,
139 *,
140 max_digits: int | None = None,
141 decimal_places: int | None = None,
142 required: bool = True,
143 allow_null: bool = False,
144 default: Any = NOT_PROVIDED,
145 validators: Sequence[Callable[..., Any]] = (),
146 ):
147 self.max_digits, self.decimal_places = max_digits, decimal_places
148 super().__init__(
149 required=required,
150 allow_null=allow_null,
151 default=default,
152 validators=validators,
153 )
154
155 def preflight(self, **kwargs: Any) -> list[PreflightResult]:
156 errors = super().preflight(**kwargs)
157
158 digits_errors = [
159 *self._check_decimal_places(),
160 *self._check_max_digits(),
161 ]
162 if not digits_errors:
163 errors.extend(self._check_decimal_places_and_max_digits())
164 else:
165 errors.extend(digits_errors)
166 return errors
167
168 def _check_decimal_places(self) -> list[PreflightResult]:
169 if self.decimal_places is None:
170 return [
171 PreflightResult(
172 fix="DecimalFields must define a 'decimal_places' attribute.",
173 obj=self,
174 id="fields.decimalfield_missing_decimal_places",
175 )
176 ]
177 try:
178 decimal_places = int(self.decimal_places)
179 if decimal_places < 0:
180 raise ValueError()
181 except ValueError:
182 return [
183 PreflightResult(
184 fix="'decimal_places' must be a non-negative integer.",
185 obj=self,
186 id="fields.decimalfield_invalid_decimal_places",
187 )
188 ]
189 else:
190 return []
191
192 def _check_max_digits(self) -> list[PreflightResult]:
193 if self.max_digits is None:
194 return [
195 PreflightResult(
196 fix="DecimalFields must define a 'max_digits' attribute.",
197 obj=self,
198 id="fields.decimalfield_missing_max_digits",
199 )
200 ]
201 try:
202 max_digits = int(self.max_digits)
203 if max_digits <= 0:
204 raise ValueError()
205 except ValueError:
206 return [
207 PreflightResult(
208 fix="'max_digits' must be a positive integer.",
209 obj=self,
210 id="fields.decimalfield_invalid_max_digits",
211 )
212 ]
213 else:
214 return []
215
216 def _check_decimal_places_and_max_digits(self) -> list[PreflightResult]:
217 if self.decimal_places is None or self.max_digits is None:
218 return []
219 if self.decimal_places > self.max_digits:
220 return [
221 PreflightResult(
222 fix="'max_digits' must be greater or equal to 'decimal_places'.",
223 obj=self,
224 id="fields.decimalfield_decimal_places_exceeds_max_digits",
225 )
226 ]
227 return []
228
229 @cached_property
230 def validators(self) -> list[Callable[..., Any]]:
231 return super().validators + [
232 validators.DecimalValidator(self.max_digits, self.decimal_places)
233 ]
234
235 @cached_property
236 def context(self) -> decimal.Context:
237 return decimal.Context(prec=self.max_digits)
238
239 def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
240 name, path, args, kwargs = super().deconstruct()
241 if self.max_digits is not None:
242 kwargs["max_digits"] = self.max_digits
243 if self.decimal_places is not None:
244 kwargs["decimal_places"] = self.decimal_places
245 return name, path, args, kwargs
246
247 def to_python(self, value: Any) -> decimal.Decimal | None:
248 if value is None:
249 return value
250 try:
251 if isinstance(value, float):
252 decimal_value = self.context.create_decimal_from_float(value)
253 else:
254 decimal_value = decimal.Decimal(value)
255 except decimal.InvalidOperation, TypeError, ValueError:
256 raise exceptions.ValidationError(
257 '"%(value)s" value must be a decimal number.',
258 code="invalid",
259 params={"value": value},
260 )
261 if not decimal_value.is_finite():
262 raise exceptions.ValidationError(
263 '"%(value)s" value must be a decimal number.',
264 code="invalid",
265 params={"value": value},
266 )
267 return decimal_value
268
269 def get_db_prep_value(
270 self, value: Any, connection: DatabaseConnection, prepared: bool = False
271 ) -> Any:
272 if not prepared:
273 value = self.get_prep_value(value)
274 return value
275
276 def get_prep_value(self, value: Any) -> Any:
277 value = super().get_prep_value(value)
278 return self.to_python(value)