"""Pydantic validation and response schema plugin for Genro Routes.
Validates handler inputs using Pydantic type hints and generates JSON Schema
from return type annotations for bridge consumption (MCP, OpenAPI).
At registration time (``on_decore``):
- Inspects parameter type hints and builds a Pydantic model for input validation.
- Inspects return type annotation and generates a JSON Schema via ``TypeAdapter``,
stored in ``entry.metadata["pydantic"]["response_schema"]``.
At call time (``wrap_handler``), validates annotated args/kwargs before calling
the real handler.
Example::
from typing import TypedDict
from genro_routes import Router, RoutingClass, route
class UserResponse(TypedDict):
id: int
name: str
class MyService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("pydantic")
@route("api")
def get_user(self, user_id: int) -> UserResponse:
return {"id": user_id, "name": "alice"}
svc = MyService()
svc.api.node("get_user")(user_id=123) # OK, validated
svc.api.node("get_user")(user_id="not_an_int") # ValidationError
# Response schema available in metadata
entry = svc.api._entries["get_user"]
entry.metadata["pydantic"]["response_schema"]
# {"type": "object", "properties": {"id": ..., "name": ...}, ...}
Configuration::
# Disable validation for a specific handler
@route("api", pydantic_disabled=True)
def unvalidated_handler(self):
pass
"""
from __future__ import annotations
import inspect
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, get_type_hints
try:
from pydantic import TypeAdapter, ValidationError, create_model
except ImportError as err: # pragma: no cover - import guard
raise ImportError(
"Pydantic plugin requires pydantic. Install with: pip install genro-routes[pydantic]"
) from err
from genro_routes.core.router import Router
from genro_routes.plugins._base_plugin import BasePlugin, MethodEntry
if TYPE_CHECKING:
from genro_routes.core import Router
[docs]
class PydanticPlugin(BasePlugin):
"""Validate handler inputs and generate response schemas with Pydantic.
At registration time (``on_decore``), builds a Pydantic model from parameter
type hints for input validation, and generates a JSON Schema from the return
type annotation via ``TypeAdapter`` for bridge consumption.
Behavior:
- Only annotated parameters are validated
- Unannotated parameters pass through unchanged
- ValidationError is raised on invalid input
- Return type annotations produce ``response_schema`` in metadata
- Can be disabled per-handler via ``pydantic_disabled=True``
Configuration options:
- ``disabled``: Skip validation for this handler/router (default False)
Attributes:
plugin_code: "pydantic" - used for registration and config prefix.
plugin_description: Human-readable description.
Example:
Basic usage::
class MyService(RoutingClass):
def __init__(self):
self.api = Router(self, name="api").plug("pydantic")
@route("api")
def get_user(self, user_id: int) -> dict[str, int]:
return {"id": user_id}
svc = MyService()
svc.api.node("get_user")(user_id=123) # OK
svc.api.node("get_user")(user_id="not_an_int") # ValidationError
# Response schema in metadata
svc.api._entries["get_user"].metadata["pydantic"]["response_schema"]
Disable validation::
@route("api", pydantic_disabled=True)
def unvalidated(self, data):
return data # no validation
"""
plugin_code = "pydantic"
plugin_description = "Validates inputs and generates response schemas using Pydantic"
[docs]
def __init__(self, router, **config: Any):
super().__init__(router, **config)
[docs]
def on_decore(self, route: Router, func: Callable, entry: MethodEntry) -> None:
"""Build Pydantic model from handler type hints and generate response schema."""
# Always capture signature info (even without type hints)
sig = inspect.signature(func)
accepts_varargs = any(
p.kind == inspect.Parameter.VAR_POSITIONAL
for p in sig.parameters.values()
)
try:
hints = get_type_hints(func, include_extras=True)
except Exception:
hints = {}
return_hint = hints.pop("return", None)
# Always save signature metadata
pydantic_meta: dict[str, Any] = {
"signature": sig,
"accepts_varargs": accepts_varargs,
"hints": hints,
}
if return_hint is not None:
pydantic_meta["return_type"] = return_hint
try:
adapter = TypeAdapter(return_hint)
pydantic_meta["response_schema"] = adapter.json_schema()
except Exception:
pass
if hints:
# Build validation model only if we have hints
fields = {}
for param_name, hint in hints.items():
param = sig.parameters.get(param_name)
if param is None:
raise ValueError(
f"Handler '{func.__name__}' has type hint for '{param_name}' "
f"which is not in the function signature"
)
elif param.default is inspect.Parameter.empty:
fields[param_name] = (hint, ...)
else:
fields[param_name] = (hint, param.default)
pydantic_meta["model"] = create_model(f"{func.__name__}_Model", **fields) # type: ignore
entry.metadata["pydantic"] = pydantic_meta
[docs]
def wrap_handler(self, route: Router, entry: MethodEntry, call_next: Callable):
"""Validate annotated parameters with the cached Pydantic model before calling."""
meta = entry.metadata.get("pydantic", {})
model = meta.get("model")
if not model:
# No model created (no type hints), passthrough
return call_next
sig = meta["signature"]
hints = meta["hints"]
def wrapper(*args, **kwargs):
# Check disabled config at runtime (not at wrap time)
cfg = self.configuration(entry.name)
if cfg.get("disabled"):
return call_next(*args, **kwargs)
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
args_to_validate = {k: v for k, v in bound.arguments.items() if k in hints}
other_args = {k: v for k, v in bound.arguments.items() if k not in hints}
try:
validated = model(**args_to_validate)
except ValidationError as exc:
raise ValidationError.from_exception_data(
title=f"Validation error in {entry.name}",
line_errors=exc.errors(),
) from exc
final_args = other_args.copy()
for key, value in validated:
final_args[key] = value
return call_next(**final_args)
return wrapper
[docs]
def get_model(self, entry: MethodEntry) -> tuple[str, Any] | None:
"""Return the Pydantic model for this handler if not disabled.
Args:
entry: The MethodEntry to get the model for.
Returns:
Tuple of ("pydantic_model", model_class) if available, else None.
"""
cfg = self.configuration(entry.name)
if cfg.get("disabled"):
return None
meta = entry.metadata.get("pydantic", {})
model = meta.get("model")
if not model:
return None
return ("pydantic_model", model)
[docs]
def entry_metadata(self, router: Any, entry: MethodEntry) -> dict[str, Any]:
"""Return pydantic metadata for introspection."""
meta = entry.metadata.get("pydantic", {})
result: dict[str, Any] = {
"model": meta.get("model"),
"hints": meta.get("hints"),
"accepts_varargs": meta.get("accepts_varargs", False),
}
response_schema = meta.get("response_schema")
if response_schema is not None:
result["response_schema"] = response_schema
return result
Router.register_plugin(PydanticPlugin)