v0.166.0
  1import datetime
  2import warnings
  3from collections.abc import Callable, Sequence
  4from functools import cached_property
  5from typing import TYPE_CHECKING, Any
  6
  7from plain.preflight import PreflightResult
  8from plain.utils import timezone
  9from plain.utils.dateparse import parse_date, parse_datetime, parse_time
 10
 11from plain import exceptions
 12
 13from .base import NOT_PROVIDED, ColumnField, DefaultableField
 14
 15if TYPE_CHECKING:
 16    from plain.postgres.base import Model
 17    from plain.postgres.connection import DatabaseConnection
 18    from plain.postgres.expressions import Func
 19
 20
 21_INVALID_DATE_MESSAGE = (
 22    '"%(value)s" value has the correct format (YYYY-MM-DD) but it is an invalid date.'
 23)
 24
 25
 26def _to_naive(value: datetime.datetime) -> datetime.datetime:
 27    if timezone.is_aware(value):
 28        value = timezone.make_naive(value, datetime.UTC)
 29    return value
 30
 31
 32def _get_naive_now() -> datetime.datetime:
 33    return _to_naive(timezone.now())
 34
 35
 36def _check_if_value_fixed(
 37    field: Any,
 38    value: datetime.date | datetime.datetime,
 39    now: datetime.datetime | None = None,
 40) -> list[PreflightResult]:
 41    """Warn if `value` looks like a fixed near-now timestamp — users usually
 42    want an `update_now=True` DateTimeField, not a one-shot literal.
 43    """
 44    if now is None:
 45        now = _get_naive_now()
 46    offset = datetime.timedelta(seconds=10)
 47    lower = now - offset
 48    upper = now + offset
 49    if isinstance(value, datetime.datetime):
 50        value = _to_naive(value)
 51    else:
 52        assert isinstance(value, datetime.date)
 53        lower = lower.date()
 54        upper = upper.date()
 55    if lower <= value <= upper:
 56        return [
 57            PreflightResult(
 58                fix="Fixed default value provided. "
 59                "It seems you set a fixed date / time / datetime "
 60                "value as default for this field. This may not be "
 61                "what you want. If you want to have the current date "
 62                "as default, use `plain.utils.timezone.now`",
 63                obj=field,
 64                id="fields.datetime_naive_default_value",
 65                warning=True,
 66            )
 67        ]
 68    return []
 69
 70
 71class DateField[T: (datetime.date, datetime.date | None) = datetime.date](
 72    DefaultableField[T]
 73):
 74    db_type_sql = "date"
 75    empty_strings_allowed = False
 76
 77    def __init__(
 78        self,
 79        *,
 80        required: bool = True,
 81        allow_null: bool = False,
 82        default: Any = NOT_PROVIDED,
 83        validators: Sequence[Callable[..., Any]] = (),
 84    ):
 85        super().__init__(
 86            required=required,
 87            allow_null=allow_null,
 88            default=default,
 89            validators=validators,
 90        )
 91
 92    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
 93        return [
 94            *super().preflight(**kwargs),
 95            *self._check_fix_default_value(),
 96        ]
 97
 98    def _check_fix_default_value(self) -> list[PreflightResult]:
 99        if not self.has_default():
100            return []
101
102        value = self.default
103        if isinstance(value, datetime.datetime):
104            value = _to_naive(value).date()
105        elif isinstance(value, datetime.date):
106            pass
107        else:
108            return []
109        return _check_if_value_fixed(self, value)
110
111    def to_python(self, value: Any) -> datetime.date | None:
112        if value is None:
113            return value
114        if isinstance(value, datetime.datetime):
115            if timezone.is_aware(value):
116                # Convert aware datetimes to the default time zone
117                # before casting them to dates (#17742).
118                default_timezone = timezone.get_default_timezone()
119                value = timezone.make_naive(value, default_timezone)
120            return value.date()
121        if isinstance(value, datetime.date):
122            return value
123
124        try:
125            parsed = parse_date(value)
126            if parsed is not None:
127                return parsed
128        except ValueError:
129            raise exceptions.ValidationError(
130                _INVALID_DATE_MESSAGE,
131                code="invalid_date",
132                params={"value": value},
133            )
134
135        raise exceptions.ValidationError(
136            '"%(value)s" value has an invalid date format. It must be in YYYY-MM-DD format.',
137            code="invalid",
138            params={"value": value},
139        )
140
141    def get_prep_value(self, value: Any) -> Any:
142        value = super().get_prep_value(value)
143        return self.to_python(value)
144
145    def get_db_prep_value(
146        self, value: Any, connection: DatabaseConnection, prepared: bool = False
147    ) -> Any:
148        if not prepared:
149            value = self.get_prep_value(value)
150        return value
151
152
153class DateTimeField[
154    T: (datetime.datetime, datetime.datetime | None) = datetime.datetime
155](ColumnField[T]):
156    db_type_sql = "timestamp with time zone"
157    empty_strings_allowed = False
158
159    def __init__(
160        self,
161        *,
162        create_now: bool = False,
163        update_now: bool = False,
164        default: Any = NOT_PROVIDED,
165        required: bool = True,
166        allow_null: bool = False,
167        validators: Sequence[Callable[..., Any]] = (),
168    ):
169        # `default` accepts only None (the nullable-optional marker, validated
170        # by ColumnField). A literal timestamp default isn't supported --
171        # create_now/update_now own the DB-side default.
172        self.create_now = create_now
173        self.update_now = update_now
174        super().__init__(
175            default=default,
176            required=required,
177            allow_null=allow_null,
178            validators=validators,
179        )
180
181    @cached_property
182    def _db_default_expression(self) -> Func | None:
183        if self.create_now:
184            from plain.postgres.functions.datetime import Now
185
186            return Now()
187        return None
188
189    def get_db_default_expression(self) -> Func | None:
190        return self._db_default_expression
191
192    @property
193    def auto_fills_on_save(self) -> bool:
194        return self.update_now
195
196    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
197        return [
198            *super().preflight(**kwargs),
199            *self._check_update_now_backfill(),
200        ]
201
202    def _check_update_now_backfill(self) -> list[PreflightResult]:
203        # update_now=True never implies a DB DEFAULT on its own (it only fires
204        # via Python pre_save), so adding the column to an existing table has
205        # no way to backfill rows unless the user also declares either a DB
206        # DEFAULT via create_now=True or a nullable column.
207        if self.update_now and not self.create_now and not self.allow_null:
208            return [
209                PreflightResult(
210                    fix="DateTimeField(update_now=True) must be paired with "
211                    "either create_now=True (DB DEFAULT backfills existing "
212                    "rows) or allow_null=True (column is nullable).",
213                    obj=self,
214                    id="fields.datetime_update_now_requires_backfill",
215                )
216            ]
217        return []
218
219    def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
220        name, path, args, kwargs = super().deconstruct()
221        if self.create_now:
222            kwargs["create_now"] = True
223        if self.update_now:
224            kwargs["update_now"] = True
225        return name, path, args, kwargs
226
227    def to_python(self, value: Any) -> datetime.datetime | None:
228        if value is None:
229            return value
230        if isinstance(value, datetime.datetime):
231            return value
232        if isinstance(value, datetime.date):
233            value = timezone.naive_datetime_from_date(value)
234
235            # For backwards compatibility, interpret naive datetimes in
236            # local time. This won't work during DST change, but we can't
237            # do much about it, so we let the exceptions percolate up the
238            # call stack.
239            warnings.warn(
240                f"DateTimeField {self.model.__name__}.{self.name} received a naive datetime "
241                f"({value}) while time zone support is active.",
242                RuntimeWarning,
243            )
244            default_timezone = timezone.get_default_timezone()
245            value = timezone.make_aware(value, default_timezone)
246
247            return value
248
249        try:
250            parsed = parse_datetime(value)
251            if parsed is not None:
252                return parsed
253        except ValueError:
254            raise exceptions.ValidationError(
255                '"%(value)s" value has the correct format (YYYY-MM-DD HH:MM[:ss[.uuuuuu]][TZ]) but it is an invalid date/time.',
256                code="invalid_datetime",
257                params={"value": value},
258            )
259
260        try:
261            parsed = parse_date(value)
262            if parsed is not None:
263                # Interpreted in the current timezone by the caller.
264                return timezone.naive_datetime_from_date(parsed)
265        except ValueError:
266            raise exceptions.ValidationError(
267                _INVALID_DATE_MESSAGE,
268                code="invalid_date",
269                params={"value": value},
270            )
271
272        raise exceptions.ValidationError(
273            '"%(value)s" value has an invalid format. It must be in YYYY-MM-DD HH:MM[:ss[.uuuuuu]][TZ] format.',
274            code="invalid",
275            params={"value": value},
276        )
277
278    def pre_save(self, model_instance: Model, add: bool) -> datetime.datetime | None:
279        if self.update_now:
280            value = timezone.now()
281            setattr(model_instance, self.name, value)
282            return value
283        return getattr(model_instance, self.name)
284
285    def get_prep_value(self, value: Any) -> Any:
286        value = super().get_prep_value(value)
287        value = self.to_python(value)
288        if value is not None and timezone.is_naive(value):
289            # For backwards compatibility, interpret naive datetimes in local
290            # time. This won't work during DST change, but we can't do much
291            # about it, so we let the exceptions percolate up the call stack.
292            try:
293                name = f"{self.model.__name__}.{self.name}"
294            except AttributeError:
295                name = "(unbound)"
296            warnings.warn(
297                f"DateTimeField {name} received a naive datetime ({value})"
298                " while time zone support is active.",
299                RuntimeWarning,
300            )
301            default_timezone = timezone.get_default_timezone()
302            value = timezone.make_aware(value, default_timezone)
303        return value
304
305    def get_db_prep_value(
306        self, value: Any, connection: DatabaseConnection, prepared: bool = False
307    ) -> Any:
308        if not prepared:
309            value = self.get_prep_value(value)
310        return value
311
312
313class TimeField[T: (datetime.time, datetime.time | None) = datetime.time](
314    DefaultableField[T]
315):
316    db_type_sql = "time without time zone"
317    empty_strings_allowed = False
318
319    def __init__(
320        self,
321        *,
322        required: bool = True,
323        allow_null: bool = False,
324        default: Any = NOT_PROVIDED,
325        validators: Sequence[Callable[..., Any]] = (),
326    ):
327        super().__init__(
328            required=required,
329            allow_null=allow_null,
330            default=default,
331            validators=validators,
332        )
333
334    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
335        return [
336            *super().preflight(**kwargs),
337            *self._check_fix_default_value(),
338        ]
339
340    def _check_fix_default_value(self) -> list[PreflightResult]:
341        if not self.has_default():
342            return []
343
344        value = self.default
345        if isinstance(value, datetime.datetime):
346            now = None
347        elif isinstance(value, datetime.time):
348            now = _get_naive_now()
349            # This will not use the right date in the race condition where now
350            # is just before the date change and value is just past 0:00.
351            value = datetime.datetime.combine(now.date(), value)
352        else:
353            return []
354        return _check_if_value_fixed(self, value, now=now)
355
356    def to_python(self, value: Any) -> datetime.time | None:
357        if value is None:
358            return None
359        if isinstance(value, datetime.time):
360            return value
361        if isinstance(value, datetime.datetime):
362            # Not usually a good idea to pass in a datetime here (it loses
363            # information), but we'll be accommodating.
364            return value.time()
365
366        try:
367            parsed = parse_time(value)
368            if parsed is not None:
369                return parsed
370        except ValueError:
371            raise exceptions.ValidationError(
372                '"%(value)s" value has the correct format (HH:MM[:ss[.uuuuuu]]) but it is an invalid time.',
373                code="invalid_time",
374                params={"value": value},
375            )
376
377        raise exceptions.ValidationError(
378            '"%(value)s" value has an invalid format. It must be in HH:MM[:ss[.uuuuuu]] format.',
379            code="invalid",
380            params={"value": value},
381        )
382
383    def get_prep_value(self, value: Any) -> Any:
384        value = super().get_prep_value(value)
385        return self.to_python(value)
386
387    def get_db_prep_value(
388        self, value: Any, connection: DatabaseConnection, prepared: bool = False
389    ) -> Any:
390        if not prepared:
391            value = self.get_prep_value(value)
392        return value