Add support for TypedDict for **kwargs

python/pydantic_core/core_schema.py

@@ -10,7 +10,7 @@
 from collections.abc import Mapping
 from datetime import date, datetime, time, timedelta
 from decimal import Decimal
-from typing import TYPE_CHECKING, Any, Callable, Dict, Hashable, List, Pattern, Set, Tuple, Type, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, Hashable, List, Pattern, Set, Tuple, Type, Union, overload
 
 from typing_extensions import deprecated
 
@@ -3372,6 +3372,48 @@ def arguments_parameter(
     return _dict_not_none(name=name, schema=schema, mode=mode, alias=alias)
 
 
+class VarKwargsSchema(TypedDict):
+    type: Literal['var_kwargs']
+    mode: Literal['single', 'typed_dict']
+    schema: CoreSchema
+
+
+@overload
+def var_kwargs_schema(
+    *,
+    mode: Literal['single'],
+    schema: CoreSchema,
+) -> VarKwargsSchema: ...
+
+
+@overload
+def var_kwargs_schema(
+    *,
+    mode: Literal['typed_dict'],
+    schema: TypedDictSchema,
+) -> VarKwargsSchema: ...
+
+
+def var_kwargs_schema(
+    *,
+    mode: Literal['single', 'typed_dict'],
+    schema: CoreSchema,
+) -> VarKwargsSchema:
+    """Returns a schema describing the variadic keyword arguments of a callable.
+
+    Args:
+        mode: The validation mode to use. If `'single'`, every value of the keyword arguments will
+            be validated against the core schema from the `schema` argument. If `'typed_dict'`, the
+            `schema` argument must be a [`typed_dict_schema`][pydantic_core.core_schema.typed_dict_schema].
+    """
+
+    return _dict_not_none(
+        type='var_kwargs',
+        mode=mode,
+        schema=schema,
+    )
+
+
 class ArgumentsSchema(TypedDict, total=False):
     type: Required[Literal['arguments']]
     arguments_schema: Required[List[ArgumentsParameter]]

src/validators/mod.rs

@@ -61,6 +61,7 @@ mod union;
 mod url;
 mod uuid;
 mod validation_state;
+mod var_kwargs;
 mod with_default;
 
 pub use self::validation_state::{Exactness, ValidationState};
@@ -561,6 +562,7 @@ pub fn build_validator(
         callable::CallableValidator,
         // arguments
         arguments::ArgumentsValidator,
+        var_kwargs::VarKwargsValidator,
         // default value
         with_default::WithDefaultValidator,
         // chain validators
@@ -716,6 +718,7 @@ pub enum CombinedValidator {
     Callable(callable::CallableValidator),
     // arguments
     Arguments(arguments::ArgumentsValidator),
+    VarKwargs(var_kwargs::VarKwargsValidator),
     // default value
     WithDefault(with_default::WithDefaultValidator),
     // chain validators

src/validators/var_kwargs.rs

@@ -0,0 +1,80 @@
+use std::str::FromStr;
+
+use pyo3::intern;
+use pyo3::prelude::*;
+use pyo3::types::{PyDict, PyString};
+
+use crate::build_tools::py_schema_err;
+use crate::errors::ValResult;
+use crate::input::Input;
+use crate::tools::SchemaDict;
+
+use super::validation_state::ValidationState;
+use super::{build_validator, BuildValidator, CombinedValidator, DefinitionsBuilder, Validator};
+
+#[derive(Debug)]
+enum VarKwargsMode {
+    Single,
+    TypedDict,
+}
+
+impl FromStr for VarKwargsMode {
+    type Err = PyErr;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "single" => Ok(Self::Single),
+            "typed_dict" => Ok(Self::TypedDict),
+            s => py_schema_err!("Invalid var_kwargs mode: `{}`, expected `single` or `typed_dict`", s),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct VarKwargsValidator {
+    mode: VarKwargsMode,
+    validator: Box<CombinedValidator>,
+}
+
+impl BuildValidator for VarKwargsValidator {
+    const EXPECTED_TYPE: &'static str = "var_kwargs";
+
+    fn build(
+        schema: &Bound<'_, PyDict>,
+        config: Option<&Bound<'_, PyDict>>,
+        definitions: &mut DefinitionsBuilder<CombinedValidator>,
+    ) -> PyResult<CombinedValidator> {
+        let py = schema.py();
+
+        let py_mode: Bound<PyString> = schema.get_as_req(intern!(py, "mode"))?;
+        let mode = VarKwargsMode::from_str(py_mode.to_string().as_str())?;
+
+        let validator_schema: Bound<PyDict> = schema.get_as_req(intern!(py, "schema"))?;
+
+        Ok(Self {
+            mode,
+            validator: Box::new(build_validator(&validator_schema, config, definitions)?),
+        }
+        .into())
+    }
+}
+
+impl_py_gc_traverse!(VarKwargsValidator { mode, validator });
+
+impl Validator for VarKwargsValidator {
+    fn validate<'py>(
+        &self,
+        py: Python<'py>,
+        input: &(impl Input<'py> + ?Sized),
+        state: &mut ValidationState<'_, 'py>,
+    ) -> ValResult<PyObject> {
+        match self.mode {
+            VarKwargsMode::Single => {
+                // TODO
+            }
+            VarKwargsMode::TypedDict => {
+                // TODO
+            }
+        }
+    }
+}