Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 45 additions & 27 deletions front/py2pydantic.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,28 @@
"""How do we eliminate some of the boilerplate between having python functions
and making them pydantic?
"""Bridge between plain Python functions and pydantic v2 models.

Given a Python function, produce a pydantic input model whose fields mirror
the function's signature (names, annotations, defaults), and an "opyrator"-
style wrapper that takes a single pydantic model instance and dispatches
to the underlying function.

Useful for auto-generating forms, validators, and JSON-schema descriptions
of arbitrary callables — the core trick behind front's app-from-function
dispatch.

>>> from i2.tests.objects_for_testing import formula1
>>> pyd_input_model = func_to_pyd_input_model_cls(formula1)
>>> pyd_input_model
<class 'pydantic.main.formula1'>
>>> pyd_input_model.__name__
'formula1'
>>> from i2 import Sig
>>> Sig(formula1)
<Sig (w, /, x: float, y=1, *, z: int = 1)>
>>> Sig(pyd_input_model)
<Sig (*, w: Any, x: float, z: int = 1, y: int = 1) -> None>
<Sig (*, w: Any, x: float, y: int = 1, z: int = 1) -> None>

>>> pyd_func = func_to_pyd_func(formula1)
>>> input_model_instance = pyd_input_model(w=1, x=2)
>>> input_model_instance
formula1(w=1, x=2.0, z=1, y=1)
formula1(w=1, x=2.0, y=1, z=1)
>>> pyd_func(input_model_instance)
3.0
>>> formula1(1, x=2) # can't say w=1 because w is position only
Expand Down Expand Up @@ -62,8 +70,7 @@ def func_to_pyd_input_model_cls(
>>> def foo(a, b: int, c: bool=False):
... ...
>>> obj = func_to_pyd_input_model_cls(foo)
>>> import json
>>> assert json.loads(obj.schema_json()) == (
>>> obj.model_json_schema() == (
... {
... 'title': 'foo',
... 'type': 'object',
Expand All @@ -74,14 +81,15 @@ def func_to_pyd_input_model_cls(
... },
... 'required': ['a', 'b']
... })
True

If some argument names of the function conflict with attribute names of BaseModel,
these will be capitalized to resolve the conflict.

>>> def bar(x, copy, schema):
... ...
>>> obj2 = func_to_pyd_input_model_cls(bar, warn_when_changing_names=False)
>>> assert json.loads(obj2.schema_json()) == (
>>> obj2.model_json_schema() == (
... {
... 'title': 'bar',
... 'type': 'object',
Expand All @@ -91,31 +99,40 @@ def func_to_pyd_input_model_cls(
... 'SCHEMA': {'title': 'Schema'}},
... 'required': ['x', 'COPY', 'SCHEMA']
... })
True

"""
name = name or name_of_obj(func)
try:
conflicting_names = set(Sig(func).names) & set(dir(BaseModel))
if not conflicting_names:
return create_model(name, **dict(func_to_pyd_model_specs(func, dflt_type)))
except NameError:
conflicting_names = set(Sig(func).names) & set(dir(BaseModel))
old_to_new_names = {k: k.upper() for k in conflicting_names}
if warn_when_changing_names:
from warnings import warn

warn(
f"""{len(conflicting_names)} argument name(s) conflicted with BaseModel.
They're being replaced with upper-case names to resolve conflict. old:new ->
{old_to_new_names}
"""
)
wrapped_func = Ingress.name_map(func, **old_to_new_names).wrap(func)
return create_model(
name, **dict(func_to_pyd_model_specs(wrapped_func, dflt_type))
# Pydantic v2 emits a UserWarning (rather than the v1 NameError) when a
# field shadows a BaseModel attribute. Detect upfront and rename so the
# resulting model is conflict-free either way.
old_to_new_names = {k: k.upper() for k in conflicting_names}
if warn_when_changing_names:
from warnings import warn

warn(
f"""{len(conflicting_names)} argument name(s) conflicted with BaseModel.
They're being replaced with upper-case names to resolve conflict. old:new ->
{old_to_new_names}
"""
)
wrapped_func = Ingress.name_map(func, **old_to_new_names).wrap(func)
return create_model(
name, **dict(func_to_pyd_model_specs(wrapped_func, dflt_type))
)


def func_to_pyd_model_specs(func: Callable, dflt_type=Any):
"""Helper function to get field info from python signature parameters"""
"""Helper function to get field info from python signature parameters.

Each spec is a ``(type, default)`` tuple suitable for pydantic v2's
``create_model``. For unannotated parameters the type is inferred from
the default value's type when a default is present, otherwise
``dflt_type`` (``Any`` by default) is used with ``...`` (required).
"""
for p in Sig(func).params:
if p.annotation is not empty_param_attr:
if p.default is not empty_param_attr:
Expand All @@ -124,7 +141,8 @@ def func_to_pyd_model_specs(func: Callable, dflt_type=Any):
yield p.name, (p.annotation, ...)
else: # no annotations
if p.default is not empty_param_attr:
yield p.name, p.default
# pydantic v2 needs an explicit type; infer from the default
yield p.name, (type(p.default), p.default)
else:
yield p.name, (dflt_type, ...)

Expand Down
36 changes: 20 additions & 16 deletions front/tests/test_py2pydantic.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
"""Unit tests for :mod:`front.py2pydantic`."""

from pydantic_core import PydanticUndefined

from front.py2pydantic import *

# ---------------------------------------------------------------------------------------
# Tests

# This one has every of the 4 combinations of (default y/n, annotated y/n)
# def formula1(w, /, x: float, y=1, *, z: int = 1):
Expand All @@ -12,20 +14,22 @@ def formula1(w, x: float, y=1, z: int = 1):
def test_func_to_pyd_model_of_inputs(func: Callable = formula1, dflt_type=Any):
pyd_model = func_to_pyd_input_model_cls(func, dflt_type)
sig = Sig(func)
# # TODO: Don't want to use hidden __fields__. Other "official" access to this info?
# the names of the arguments should correspond to names of the model's fields:
assert sorted(sig.names) == sorted(pyd_model.__fields__)
# and for each field, name, default, and sometimes annotations/types should match...
for name, model_field in pyd_model.__fields__.items():
# The function arg names should match the model's field names.
assert sorted(sig.names) == sorted(pyd_model.model_fields)
# And for each field, default and annotation should match the signature.
for name, model_field in pyd_model.model_fields.items():
param = sig.parameters[name]
expected_default = (
param.default if param.default is not empty_param_attr else None
)
assert model_field.default == expected_default
if param.default is empty_param_attr:
# No default → pydantic marks the field required (PydanticUndefined sentinel).
assert model_field.default is PydanticUndefined
assert model_field.is_required()
else:
assert model_field.default == param.default
assert not model_field.is_required()
if param.annotation is not empty_param_attr:
# if arg is annotated, expect that as the field type
assert model_field.type_ == param.annotation
# If arg is annotated, expect that as the field type.
assert model_field.annotation == param.annotation
elif param.default is empty_param_attr:
# if arg is not annotated and doesn't have a default, expect dflt_type
assert model_field.type_ == dflt_type
# but don't test pydantic's resolution of types from default values
# If arg is not annotated and has no default, expect dflt_type.
assert model_field.annotation == dflt_type
# else: pydantic infers the type from the default's type — we don't pin that here.
108 changes: 108 additions & 0 deletions front/tests/test_py2pydantic_integration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""Integration tests for the function → pydantic-model → wrapped-call pipeline.

The unit test in :mod:`test_py2pydantic` only checks the *structure* of the
generated input model. This file checks the **end-to-end dispatch path** that
front and streamlitfront depend on:

- generate a pydantic input model from a Python function
- instantiate it with values (this exercises pydantic's coercion/validation)
- feed it into the wrapped function and confirm the original function is
invoked correctly and returns the expected result
- confirm validation errors surface as ``ValidationError`` instead of
silently passing through

If this file goes red after a pydantic version bump or signature change,
something has broken the contract that downstream front-based dispatch
relies on.
"""

import pytest
from pydantic import ValidationError

from front.py2pydantic import (
func_to_pyd_input_model_cls,
func_to_pyd_func,
pydantic_model_from_type,
)


# Three of the four parameter kinds — the fourth (unannotated, default=None)
# is intentionally avoided here because pydantic v2 infers NoneType from a
# None default, which is a surprising-by-design behavior unrelated to dispatch.
def example(a: int, b: float = 1.5, d=10):
return {"a": a, "b": b, "d": d, "sum": a + b + d}


def test_input_model_roundtrip_dispatch():
"""Build the input model, instantiate it, dispatch through the wrapper."""
InputModel = func_to_pyd_input_model_cls(example)
pyd_example = func_to_pyd_func(example)

instance = InputModel(a=3, b=2.5)
result = pyd_example(instance)

assert result == {"a": 3, "b": 2.5, "d": 10, "sum": 15.5}


def test_input_model_coerces_types():
"""Pydantic should coerce a stringified int into an int per the annotation."""
InputModel = func_to_pyd_input_model_cls(example)
# "5" is coerced to int 5 by pydantic (lax mode is the v2 default)
instance = InputModel(a="5", b=0.0)
assert instance.a == 5
assert isinstance(instance.a, int)


def test_input_model_raises_on_invalid_type():
"""Pydantic should raise ValidationError when input can't be coerced."""
InputModel = func_to_pyd_input_model_cls(example)
with pytest.raises(ValidationError):
InputModel(a="not-an-int", b=0.0)


def test_input_model_missing_required():
"""Missing a required field surfaces as ValidationError, not a silent pass."""
InputModel = func_to_pyd_input_model_cls(example)
with pytest.raises(ValidationError):
InputModel(b=0.0) # missing `a`


def test_input_model_json_schema_shape():
"""The schema reflects the function signature's contract."""
InputModel = func_to_pyd_input_model_cls(example)
schema = InputModel.model_json_schema()
assert schema["type"] == "object"
# `a` is required, `b`/`c`/`d` have defaults
assert "a" in schema["required"]
assert "b" not in schema.get("required", [])
# Annotated params carry their declared type
assert schema["properties"]["a"]["type"] == "integer"
assert schema["properties"]["b"]["type"] == "number"


def test_output_model_round_trip():
"""`pydantic_model_from_type` produces a model that wraps the chosen type."""
OutputModel = pydantic_model_from_type(int, name="MyOutput", field_name="value")
inst = OutputModel(value=42)
assert inst.value == 42
with pytest.raises(ValidationError):
OutputModel(value="not-an-int-or-coercible")


def test_basemodel_name_conflict_handled():
"""Param names that collide with BaseModel attrs get auto-uppercased.

Without this, pydantic emits a UserWarning *and* shadows the BaseModel
attribute. The renaming keeps the model class clean.
"""
import warnings

def f(x, copy, schema):
return (x, copy, schema)

with warnings.catch_warnings():
warnings.simplefilter("ignore")
InputModel = func_to_pyd_input_model_cls(f, warn_when_changing_names=False)

fields = set(InputModel.model_fields)
assert fields == {"x", "COPY", "SCHEMA"}
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ dependencies = [
"dol",
"i2",
"meshed",
"pydantic==1.10.12",
"pydantic>=2",
]

[project.license]
Expand Down
Loading