Implement datetime timezone constraints

This implements support for constraining datetime objects based on them
having or not having timezone info. The aware kind constrains to objects
that have timezone info, and symmetrically, the naive kind constrains to
objects that do not have timezone info.

In addition to testing the constraints individually, a test is added
to assert the property that all datetime objects are either aware or
naive, but never both.

Author: antonagestam

pydantic_core/core_schema.py

@@ -276,6 +276,7 @@ class DatetimeSchema(TypedDict, total=False):
     lt: datetime
     gt: datetime
     now_op: Literal['past', 'future']
+    tz_constraint: Literal['aware', 'naive']
     # defaults to current local utc offset from `time.localtime().tm_gmtoff`
     # value is restricted to -86_400 < offset < 86_400 by bounds in generate_self_schema.py
     now_utc_offset: int
@@ -291,6 +292,7 @@ def datetime_schema(
     lt: datetime | None = None,
     gt: datetime | None = None,
     now_op: Literal['past', 'future'] | None = None,
+    tz_constraint: Literal['aware', 'naive'] | None = None,
     now_utc_offset: int | None = None,
     ref: str | None = None,
     extra: Any = None,
@@ -303,6 +305,7 @@ def datetime_schema(
         lt=lt,
         gt=gt,
         now_op=now_op,
+        tz_constraint=tz_constraint,
         now_utc_offset=now_utc_offset,
         ref=ref,
         extra=extra,
@@ -1226,6 +1229,8 @@ def multi_host_url_schema(
     'datetime_object_invalid',
     'datetime_past',
     'datetime_future',
+    'datetime_aware',
+    'datetime_naive',
     'time_delta_type',
     'time_delta_parsing',
     'frozen_set_type',

src/errors/types.rs

@@ -261,6 +261,10 @@ pub enum ErrorType {
     DatetimePast,
     #[strum(message = "Datetime should be in the future")]
     DatetimeFuture,
+    #[strum(message = "Datetime should have timezone info")]
+    DatetimeAware,
+    #[strum(message = "Datetime should not have timezone info")]
+    DatetimeNaive,
     // ---------------------
     // timedelta errors
     #[strum(message = "Input should be a valid timedelta")]

src/validators/datetime.rs

@@ -92,6 +92,14 @@ impl Validator for DateTimeValidator {
                     }
                 }
             }
+
+            if let Some(ref tz_constraint) = constraints.tz {
+                match (&tz_constraint.kind, speedate_dt.offset) {
+                    (TZConstraintKind::Aware, None) => return Err(ValError::new(ErrorType::DatetimeAware, input)),
+                    (TZConstraintKind::Naive, Some(_)) => return Err(ValError::new(ErrorType::DatetimeNaive, input)),
+                    _ => (),
+                }
+            }
         }
         Ok(datetime.try_into_py(py)?)
     }
@@ -108,6 +116,7 @@ struct DateTimeConstraints {
     ge: Option<DateTime>,
     gt: Option<DateTime>,
     now: Option<NowConstraint>,
+    tz: Option<TZConstraint>,
 }
 
 impl DateTimeConstraints {
@@ -119,8 +128,9 @@ impl DateTimeConstraints {
             ge: py_datetime_as_datetime(schema, intern!(py, "ge"))?,
             gt: py_datetime_as_datetime(schema, intern!(py, "gt"))?,
             now: NowConstraint::from_py(schema)?,
+            tz: TZConstraint::from_py(schema)?,
         };
-        if c.le.is_some() || c.lt.is_some() || c.ge.is_some() || c.gt.is_some() || c.now.is_some() {
+        if c.le.is_some() || c.lt.is_some() || c.ge.is_some() || c.gt.is_some() || c.now.is_some() || c.tz.is_some() {
             Ok(Some(c))
         } else {
             Ok(None)
@@ -199,3 +209,37 @@ impl NowConstraint {
         }
     }
 }
+
+#[derive(Debug, Clone)]
+pub enum TZConstraintKind {
+    Aware,
+    Naive,
+}
+
+impl TZConstraintKind {
+    pub fn from_str(s: &str) -> PyResult<Self> {
+        match s {
+            "aware" => Ok(TZConstraintKind::Aware),
+            "naive" => Ok(TZConstraintKind::Naive),
+            _ => py_err!("Invalid tz_constraint {:?}", s),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct TZConstraint {
+    pub kind: TZConstraintKind,
+}
+
+impl TZConstraint {
+    pub fn from_py(schema: &PyDict) -> PyResult<Option<Self>> {
+        let py = schema.py();
+
+        match schema.get_as(intern!(py, "tz_constraint"))? {
+            Some(kind) => Ok(Some(Self {
+                kind: TZConstraintKind::from_str(kind)?,
+            })),
+            None => Ok(None),
+        }
+    }
+}

tests/test_errors.py

@@ -235,6 +235,8 @@ def f(input_value, **kwargs):
     ('datetime_object_invalid', 'Invalid datetime object, got foobar', {'error': 'foobar'}),
     ('datetime_past', 'Datetime should be in the past', None),
     ('datetime_future', 'Datetime should be in the future', None),
+    ('datetime_aware', 'Datetime should have timezone info', None),
+    ('datetime_naive', 'Datetime should not have timezone info', None),
     ('time_delta_type', 'Input should be a valid timedelta', None),
     ('time_delta_parsing', 'Input should be a valid timedelta, foobar', {'error': 'foobar'}),
     ('frozen_set_type', 'Input should be a valid frozenset', None),

tests/validators/test_datetime.py

@@ -6,6 +6,8 @@
 
 import pytest
 import pytz
+from hypothesis import example, given
+from hypothesis.strategies import datetimes, timezones
 
 from pydantic_core import SchemaError, SchemaValidator, ValidationError, core_schema
 
@@ -366,3 +368,60 @@ def test_mock_utc_offset_8_hours(mocker):
 def test_offset_too_large():
     with pytest.raises(SchemaError, match=r'Input should be greater than -86400 \[type=greater_than,'):
         SchemaValidator(core_schema.datetime_schema(now_op='past', now_utc_offset=-24 * 3600))
+
+
+class TestTZConstraints:
+    aware_validator = SchemaValidator(core_schema.datetime_schema(tz_constraint='aware'))
+    naive_validator = SchemaValidator(core_schema.datetime_schema(tz_constraint='naive'))
+
+    def test_raises_schema_error_for_unknown_constraint_kind(self):
+        with pytest.raises(
+            SchemaError,
+            match=(
+                r'Input should be \'aware\' or \'naive\' '
+                r'\[type=literal_error, input_value=\'foo\', input_type=str\]'
+            ),
+        ):
+            SchemaValidator({'type': 'datetime', 'tz_constraint': 'foo'})
+
+    @given(datetimes(timezones=timezones()))
+    @example(datetime.now(tz=timezone.utc))
+    def test_can_validate_aware_value(self, value: datetime):
+        assert value is self.aware_validator.validate_python(value)
+
+    @given(datetimes())
+    @example(datetime.now())
+    def test_raises_validation_error_when_aware_given_naive(self, value: datetime):
+        with pytest.raises(ValidationError, match=r'Datetime should have timezone info'):
+            assert self.aware_validator.validate_python(value)
+
+    @given(datetimes())
+    @example(datetime.now())
+    def test_can_validate_naive_value(self, value: datetime):
+        assert value is self.naive_validator.validate_python(value)
+
+    @given(datetimes(timezones=timezones()))
+    @example(datetime.now(tz=timezone.utc))
+    def test_raises_validation_error_when_naive_given_aware(self, value: datetime):
+        with pytest.raises(ValidationError, match=r'Datetime should not have timezone info'):
+            assert self.naive_validator.validate_python(value)
+
+    @given(datetimes() | datetimes(timezones=timezones()))
+    @example(datetime.now())
+    @example(datetime.now(tz=timezone.utc))
+    def test_aware_and_naive_are_mutually_exclusive(self, value: datetime):
+        try:
+            self.aware_validator.validate_python(value)
+        except ValidationError:
+            is_aware = False
+        else:
+            is_aware = True
+
+        try:
+            self.naive_validator.validate_python(value)
+        except ValidationError:
+            is_naive = False
+        else:
+            is_naive = True
+
+        assert (is_aware or is_naive) and not (is_aware and is_naive)