implement never

Author: changhc

python/pydantic_core/core_schema.py

@@ -3858,6 +3858,46 @@ def definition_reference_schema(
     )
 
 
+class NeverSchema(TypedDict, total=False):
+    type: Required[Literal['never']]
+    ref: str
+    metadata: Dict[str, Any]
+
+
+def never_schema(
+    *,
+    ref: str | None = None,
+    metadata: Dict[str, Any] | None = None,
+) -> NeverSchema:
+    """
+    Returns a schema that represents a `typing.Never` field, e.g.:
+
+    ```py
+    from pydantic_core import SchemaValidator, core_schema
+
+    schema = core_schema.never_schema()
+    v = SchemaValidator(schema) # should always fail
+    try:
+        assert v.validate_python(1)
+    except ValidationError:
+        pass
+    try:
+        assert v.validate_python('s')
+    except ValidationError:
+        pass
+    ```
+
+    Args:
+        ref: optional unique identifier of the schema, used to reference the schema in other places
+        metadata: Any other information you want to include with the schema, not used by pydantic-core
+    """
+    return _dict_not_none(
+        type='never',
+        ref=ref,
+        metadata=metadata,
+    )
+
+
 MYPY = False
 # See https://github.com/python/mypy/issues/14034 for details, in summary mypy is extremely slow to process this
 # union which kills performance not just for pydantic, but even for code using pydantic
@@ -3913,6 +3953,7 @@ def definition_reference_schema(
         DefinitionReferenceSchema,
         UuidSchema,
         ComplexSchema,
+        NeverSchema,
     ]
 elif False:
     CoreSchema: TypeAlias = Mapping[str, Any]
@@ -3970,6 +4011,7 @@ def definition_reference_schema(
     'definition-ref',
     'uuid',
     'complex',
+    'never',
 ]
 
 CoreSchemaFieldType = Literal['model-field', 'dataclass-field', 'typed-dict-field', 'computed-field']

src/errors/types.rs

@@ -430,6 +430,7 @@ error_types! {
     // Complex errors
     ComplexType {},
     ComplexStrParsing {},
+    Never {},
 }
 
 macro_rules! render {
@@ -576,6 +577,7 @@ impl ErrorType {
             Self::DecimalWholeDigits {..} => "Decimal input should have no more than {whole_digits} digit{expected_plural} before the decimal point",
             Self::ComplexType {..} => "Input should be a valid python complex object, a number, or a valid complex string following the rules at https://docs.python.org/3/library/functions.html#complex",
             Self::ComplexStrParsing {..} => "Input should be a valid complex string following the rules at https://docs.python.org/3/library/functions.html#complex",
+            Self::Never { .. } => "Unexpected input for a field that should never be filled"
         }
     }
 

src/serializers/shared.rs

@@ -143,6 +143,7 @@ combined_serializer! {
         Recursive: super::type_serializers::definitions::DefinitionRefSerializer;
         Tuple: super::type_serializers::tuple::TupleSerializer;
         Complex: super::type_serializers::complex::ComplexSerializer;
+        Never: super::type_serializers::never::NeverSerializer;
     }
 }
 
@@ -254,6 +255,7 @@ impl PyGcTraverse for CombinedSerializer {
             CombinedSerializer::Tuple(inner) => inner.py_gc_traverse(visit),
             CombinedSerializer::Uuid(inner) => inner.py_gc_traverse(visit),
             CombinedSerializer::Complex(inner) => inner.py_gc_traverse(visit),
+            CombinedSerializer::Never(inner) => inner.py_gc_traverse(visit),
         }
     }
 }

src/serializers/type_serializers/mod.rs

@@ -16,6 +16,7 @@ pub mod json_or_python;
 pub mod list;
 pub mod literal;
 pub mod model;
+pub mod never;
 pub mod nullable;
 pub mod other;
 pub mod set_frozenset;

src/serializers/type_serializers/model.rs

@@ -62,7 +62,12 @@ impl BuildSerializer for ModelFieldsBuilder {
                 let serializer = CombinedSerializer::build(&schema, config, definitions)
                     .map_err(|e| py_schema_error_type!("Field `{}`:\n  {}", key, e))?;
 
-                fields.insert(key, SerField::new(py, key_py, alias, Some(serializer), true));
+                match serializer {
+                    CombinedSerializer::Never(_) => {}
+                    s => {
+                        fields.insert(key, SerField::new(py, key_py, alias, Some(s), true));
+                    }
+                }
             }
         }
 

src/serializers/type_serializers/never.rs

@@ -0,0 +1,57 @@
+use super::{py_err_se_err, BuildSerializer, CombinedSerializer, Extra, TypeSerializer};
+use crate::definitions::DefinitionsBuilder;
+use crate::tools::py_err;
+use pyo3::exceptions::PyTypeError;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+use std::borrow::Cow;
+
+const ERROR_MESSAGE: &str = "type `never` cannot be serialized";
+
+#[derive(Debug)]
+pub struct NeverSerializer;
+
+impl BuildSerializer for NeverSerializer {
+    const EXPECTED_TYPE: &'static str = "never";
+
+    fn build(
+        _schema: &Bound<'_, PyDict>,
+        _config: Option<&Bound<'_, PyDict>>,
+        _definitions: &mut DefinitionsBuilder<CombinedSerializer>,
+    ) -> PyResult<CombinedSerializer> {
+        Ok(Self {}.into())
+    }
+}
+
+impl_py_gc_traverse!(NeverSerializer {});
+
+impl TypeSerializer for NeverSerializer {
+    fn to_python(
+        &self,
+        _value: &Bound<'_, PyAny>,
+        _include: Option<&Bound<'_, PyAny>>,
+        _exclude: Option<&Bound<'_, PyAny>>,
+        _extra: &Extra,
+    ) -> PyResult<PyObject> {
+        py_err!(PyTypeError; ERROR_MESSAGE)
+    }
+
+    fn json_key<'a>(&self, _key: &'a Bound<'_, PyAny>, _extra: &Extra) -> PyResult<Cow<'a, str>> {
+        py_err!(PyTypeError; ERROR_MESSAGE)
+    }
+
+    fn serde_serialize<S: serde::ser::Serializer>(
+        &self,
+        _value: &Bound<'_, PyAny>,
+        _serializer: S,
+        _include: Option<&Bound<'_, PyAny>>,
+        _exclude: Option<&Bound<'_, PyAny>>,
+        _extra: &Extra,
+    ) -> Result<S::Ok, S::Error> {
+        py_err!(PyTypeError; ERROR_MESSAGE).map_err(py_err_se_err)
+    }
+
+    fn get_name(&self) -> &str {
+        Self::EXPECTED_TYPE
+    }
+}

src/validators/mod.rs

@@ -49,6 +49,7 @@ mod list;
 mod literal;
 mod model;
 mod model_fields;
+mod never;
 mod none;
 mod nullable;
 mod set;
@@ -611,6 +612,7 @@ pub fn build_validator(
         definitions::DefinitionRefValidator,
         definitions::DefinitionsValidatorBuilder,
         complex::ComplexValidator,
+        never::NeverValidator,
     )
 }
 
@@ -765,6 +767,7 @@ pub enum CombinedValidator {
     // input dependent
     JsonOrPython(json_or_python::JsonOrPython),
     Complex(complex::ComplexValidator),
+    Never(never::NeverValidator),
 }
 
 /// This trait must be implemented by all validators, it allows various validators to be accessed consistently,

src/validators/never.rs

@@ -0,0 +1,60 @@
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+
+use crate::errors::{ErrorTypeDefaults, ValError, ValResult};
+use crate::input::Input;
+use crate::PydanticUndefinedType;
+
+use super::{BuildValidator, CombinedValidator, DefinitionsBuilder, LocItem, ValidationState, Validator};
+
+#[derive(Debug)]
+pub struct NeverValidator {
+    undefined: PyObject,
+}
+
+impl BuildValidator for NeverValidator {
+    const EXPECTED_TYPE: &'static str = "never";
+
+    fn build(
+        schema: &Bound<'_, PyDict>,
+        _config: Option<&Bound<'_, PyDict>>,
+        _definitions: &mut DefinitionsBuilder<CombinedValidator>,
+    ) -> PyResult<CombinedValidator> {
+        let py = schema.py();
+        Ok(Self {
+            undefined: PydanticUndefinedType::new(py).to_object(py),
+        }
+        .into())
+    }
+}
+
+impl_py_gc_traverse!(NeverValidator {});
+
+impl Validator for NeverValidator {
+    fn validate<'py>(
+        &self,
+        py: Python<'py>,
+        input: &(impl Input<'py> + ?Sized),
+        _state: &mut ValidationState<'_, 'py>,
+    ) -> ValResult<PyObject> {
+        let obj = input.to_object(py);
+        if obj.is(&self.undefined) {
+            Ok(obj)
+        } else {
+            Err(ValError::new(ErrorTypeDefaults::Never, input))
+        }
+    }
+
+    fn default_value<'py>(
+        &self,
+        _py: Python<'py>,
+        _outer_loc: Option<impl Into<LocItem>>,
+        _state: &mut ValidationState<'_, 'py>,
+    ) -> ValResult<Option<PyObject>> {
+        Ok(Some(self.undefined.clone()))
+    }
+
+    fn get_name(&self) -> &str {
+        Self::EXPECTED_TYPE
+    }
+}

tests/serializers/test_model.py

@@ -3,7 +3,7 @@
 import platform
 import warnings
 from random import randint
-from typing import Any, ClassVar, Dict
+from typing import Any, ClassVar, Dict, Never
 
 try:
     from functools import cached_property
@@ -1152,3 +1152,25 @@ class BModel(BasicModel): ...
     with pytest.warns(UserWarning, match='Expected 2 fields but got 1 for type `.*AModel` with value `.*`.+'):
         value = BasicModel(root=AModel(type='a'))
         s.to_python(value)
+
+
+def test_never():
+    class MyModel:
+        a: int
+        b: Never
+
+    schema = core_schema.model_schema(
+        MyModel,
+        core_schema.model_fields_schema(
+            {
+                'a': core_schema.model_field(core_schema.int_schema()),
+                'b': core_schema.model_field(core_schema.never_schema()),
+            }
+        ),
+    )
+    v = SchemaValidator(schema)
+    m = v.validate_python({'a': 1})
+    s = SchemaSerializer(schema)
+    # `b` should not break the serialiser or be serialised
+    assert s.to_python(m) == {'a': 1}
+    assert json.loads(s.to_json(m)) == {'a': 1}

tests/serializers/test_never.py

@@ -0,0 +1,17 @@
+import pytest
+
+from pydantic_core import PydanticSerializationError, SchemaSerializer, core_schema
+
+
+def test_to_python_never():
+    v = SchemaSerializer(core_schema.never_schema())
+    with pytest.raises(TypeError) as exc_info:
+        v.to_python(1)
+    assert str(exc_info.value) == 'type `never` cannot be serialized'
+
+
+def test_to_json_never():
+    v = SchemaSerializer(core_schema.never_schema())
+    with pytest.raises(PydanticSerializationError) as exc_info:
+        v.to_json('null')
+    assert 'type `never` cannot be serialized' in str(exc_info.value)