2A module that implements the Schema interface using pydantic.
9from pydantic
import BaseModel, validate_call, ConfigDict, Field
11from formatron.schemas.schema
import FieldInfo, Schema
16 A wrapper for pydantic FieldInfo.
18 __slots__ = (
"_field",)
20 def __init__(self, field: pydantic.fields.FieldInfo):
22 Initialize the field information.
27 def annotation(self) -> typing.Type[typing.Any] | None:
28 return self.
_field.annotation
32 return self.
_field.is_required()
43 A wrapper for pydantic BaseModel that implements the Schema interface.
45 __cached_fields__ =
None
48 def fields(cls) -> typing.Dict[str, typing.Any]:
49 if cls.__cached_fields__
is not None:
50 return cls.__cached_fields__
51 cls.__cached_fields__ = {k:
FieldInfo(v)
for k, v
in cls.model_fields.items()}
52 return cls.__cached_fields__
55 def from_json(cls, _json: str) ->
"ClassSchema":
57 Create a ClassSchema from a JSON string.
59 return cls.model_validate_json(_json)
62CallableT = typing.TypeVar(
'CallableT', bound=typing.Callable)
65def callable_schema(func: CallableT, /, *, config: ConfigDict =
None, validate_return: bool =
False) -> CallableT:
67 A decorator that wraps pydantic's validate_call. The decorated callable also implements the Schema interface.
70 func: The function to decorate.
71 config: The pydantic configuration of validate_call.
72 validate_return: Whether to validate the return value.
75 The decorated callable.
77 pydantic_wrapper = validate_call(config=config, validate_return=validate_return)(func)
78 signature = inspect.signature(func, eval_str=
True)
80 for k, p
in signature.parameters.items():
82 if p.default
is not inspect.Signature.empty:
84 actual_type = p.annotation
86 if isinstance(p.default, pydantic.fields.FieldInfo):
88 if typing.get_origin(p.annotation)
is typing.Annotated:
89 actual_type, *meta = typing.get_args(p.annotation)
92 if isinstance(i, pydantic.fields.FieldInfo):
96 if fieldinfo
is not None:
99 fields[k].default = default
100 fields[k].annotation = actual_type
101 fields[k].metadata.extend(metadata)
103 if default
is not None:
104 fields[k] = Field(default)
107 fields[k].annotation = actual_type
108 fields[k].metadata.extend(metadata)
112 def from_json(cls, json_str):
113 json_data = json.loads(json_str)
116 for k, p
in signature.parameters.items():
117 if p.kind == p.POSITIONAL_ONLY:
118 positional_only.append(json_data[k])
120 others[k] = json_data[k]
121 return cls(*positional_only, **others)
124 f
'{func.__qualname__}_PydanticSchema_{id(func)}',
127 "_func": pydantic_wrapper,
128 '__new__':
lambda cls, *args, **kwargs: pydantic_wrapper(*args, **kwargs),
129 '__call__':
lambda *args, **kwargs: pydantic_wrapper(*args, **kwargs)
132 _class.from_json = classmethod(from_json)
133 _class.fields = classmethod(
lambda cls: fields)