Functional Form for data & tests

pydantic_form/form.py

@@ -1,7 +1,8 @@
 from collections import defaultdict
 from wtforms import Form, FieldList, FormField, Field
 from pydantic import ValidationError, BaseModel
-import types
+
+from .translator import SchemaToForm
 
 
 class PydanticFieldList(FieldList):
@@ -22,21 +23,23 @@ class PydanticForm(Form):
     _baked_instance = None
 
     def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
         self._errors = {}
+        self.translator = SchemaToForm(self._schema, self)
+        super().__init__(*args, **kwargs)
 
     def process(self, formdata=None, obj=None, data=None, **kwargs):
+        super().process()
         formdata = self.meta.wrap_formdata(self, formdata)
         input = formdata or data or kwargs
-        try:
-            valid = self._schema(**input)
-            for k, v in valid.dict().items():
-                setattr(self, k, v)
-        except ValidationError as e:
-            self._errors = e.errors()
-
-        for k, v in self._fields.items():
-            ...#setattr(getattr(self, k), 'data', None)
+        self.translator(input)
+        self.translator.set_data()
 
     def validate(self):
+        if self.translator.errors is not None:
+            self.translator.set_errors()
+            return False
+
+        return True
+
+    def process_obj(self, obj):
         ...

pydantic_form/iterators.py

@@ -0,0 +1,54 @@
+from wtforms import Form, FormField
+
+
+def iter_field(parent_class, leafs_only=True, path=()):
+    field = getattr(parent_class, path[-1]) if path else parent_class
+
+    if issubclass(field.field_class, FormField):
+        field_class = field.kwargs['form_class']
+        if path and not leafs_only:
+            yield path, field_class
+
+        for subfield_name, subfield in field_class._unbound_fields:
+            yield from iter_field(field_class, leafs_only, path + (subfield_name,))
+
+    else:
+        yield path, field
+
+
+def iter_form_class(form_class, leafs_only=True, path=()):
+    field = getattr(form_class, path[-1]) if path else form_class
+
+    if path and not leafs_only:
+        yield path, field
+
+    try:
+        for subfield_name, subfield in field._unbound_fields:
+            yield from iter_field(field, leafs_only, path + (subfield_name,))
+    except TypeError as e:
+        # Ensure that the _unbound_fields is populated (that happens on first
+        # instantiation)
+        form_class()
+        yield from iter_form_class(field, leafs_only, path )
+
+def iter_form(form, leafs_only=True, path=()):
+    field = getattr(form, path[-1]) if path else form
+    if isinstance(field, Form) or isinstance(field, FormField):
+        if path and not leafs_only:
+            yield path, field
+
+        for f in field._fields:
+            yield from iter_form(field, leafs_only, path + (f,))
+    else:
+        yield path, field
+
+def iter_schema(schema, leafs_only=True, path=()):
+    type_ = schema.type_ if path else schema
+    if hasattr(type_, '__fields__'):
+        if path and not leafs_only:
+            yield path, schema
+
+        for key, value in type_.__fields__.items():
+            yield from iter_schema(value, leafs_only, path + (key,))
+    else:
+        yield path, schema

pydantic_form/translator.py

@@ -1,65 +1,77 @@
-from typing import Any
+from collections import defaultdict, abc
+from .iterators import *
 from .utils import *
-
-
-class FieldTranslator:
-    def __init__(self, src, dest):
-        self.src = src
-        self.dest = dest
-
-    def get(self, key: tuple) -> Any:
-        raise NotImplementedError()
-
-    def transform(self, value: Any) -> Any:
-        raise NotImplementedError()
-
-    def set(self, src_key, value) -> None:
-        raise NotImplementedError()
-
-    def __call__(self, key):
-        return self.set(key, self.transform(self.get(key)))
-
-
-class InstanceTranslator:
-    field_translator_classes = (
-        ('valid', None),
-    )
-
-    def __init__(self, src, dest, keys):
-        self.field_translators = [
-            (k, v(src, dest)) for k, v in self.field_translator_classes
-        ]
-        self.keys = keys
-
-    def __call__(self, *args, **kwargs):
-        for k in self.keys:
-            for _, t in self.field_translators:
-                t(k)
-
-
-class SchemaInstanceToFormDataField(FieldTranslator):
-    def get(self, key: tuple):
-        return recursive_get(self.src.dict(), *key)
-
-    def transform(self, value: Any):
-        return value
-
-    def set(self, src_key, value):
-        rsetattr(self.dest, src_key + ('data',), value)
-
-
-class SchemaInstanceToFormErrorField(FieldTranslator):
-    def get(self, key: tuple):
-        return recursive_get(self.src.dict(), *key)
-
-    def transform(self, value: Any):
-        return value
-
-    def set(self, src_key, value):
-        rsetattr(self.dest, src_key + ('data',), value)
-
-
-class SchemaToForm(InstanceTranslator):
-    field_translator_classes = [
-        ('valid', SchemaInstanceToFormDataField),
-    ]
+from pydantic import ValidationError
+
+
+def default_to_regular(d):
+    if isinstance(d, defaultdict):
+        d = {k: default_to_regular(v) for k, v in d.items()}
+    return d
+
+
+def nested_dict_iter(nested, path=()):
+    for key, value in nested.items():
+        if isinstance(value, abc.Mapping):
+            yield from nested_dict_iter(value, path + (key,))
+        else:
+            yield path + (key,), value
+
+
+class SchemaToForm:
+    _schema = None
+    _errors = None
+
+    def __init__(self, schema_class, form, lut=None):
+        self.schema_class = schema_class
+        self.form = form
+        self.lut = lut if lut else lambda x: x
+
+    def __call__(self, data):
+        try:
+            self._schema = self.schema_class(**data)
+            self._errors = None
+        except ValidationError as e:
+            self._errors = defaultdict(list)
+            for error in e.errors():
+                light_error = {k: error[k] for k in error if k != 'loc'}
+                recursive_dict_operation(
+                    self._errors,
+                    lambda d, k: d.setdefault(k, defaultdict(list)),
+                    lambda d, k: d[k].append(light_error),
+                    *error['loc']
+                )
+
+            self._errors = default_to_regular(self._errors)
+            self._schema = self.schema_class.construct(data, self.schema_class.__fields__)
+
+    @property
+    def schema(self):
+        return self._schema
+
+    @property
+    def errors(self):
+        return self._errors
+
+    def set_data(self):
+        schema = self._schema
+        for src_key, src_value in iter_schema(schema):
+            try:
+                value = rgetattr(schema, src_key)
+            except AttributeError:
+                try:
+                    value = recursive_get(getattr(schema, src_key[0]), *src_key[1:])
+                except AttributeError:
+                    continue
+
+            dest_key = self.lut(src_key)
+            dest_field = rgetattr(self.form, dest_key)
+            setattr(dest_field, 'data', value)
+
+    def set_errors(self):
+        for k, error_list in nested_dict_iter(self.errors):
+            field = rgetattr(self.form, k)
+            if isinstance(field, FormField):
+                setattr(field.form, '_errors', error_list)
+            else:
+                setattr(field, 'errors', error_list)

pydantic_form/utils.py

@@ -13,18 +13,35 @@ def recursive_setdefault(root, value, *keys):
     inner.update({keys[-1]: value})
     return root
 
+def recursive_dict_operation(root, default_fn, operation_fn, *keys):
+    """https://stackoverflow.com/a/21025122/8783170"""
+    inner = functools.reduce(default_fn, keys[:-1], root)
+    operation_fn(inner, keys[-1])
+    return root
 
-def rsetattr(obj, attr, val):
+
+def rsetattr_str(obj, attr, val):
     """https://stackoverflow.com/a/31174427/8783170"""
-    attr = '.'.join(attr)
     pre, _, post = attr.rpartition('.')
-    return setattr(rgetattr(obj, pre) if pre else obj, post, val)
+    return setattr(rgetattr_str(obj, pre) if pre else obj, post, val)
 
 
-def rgetattr(obj, attr, *args):
+def rgetattr_str(obj, attr, *args):
+    """https://stackoverflow.com/a/31174427/8783170"""
+
+    def _getattr(obj, attr):
+        return getattr(obj, attr, *args)
+
+    return functools.reduce(_getattr, [obj] + attr.split('.'))
+
+def rsetattr(obj, attr, val):
     """https://stackoverflow.com/a/31174427/8783170"""
+    *pre, post = attr
+    return setattr(rgetattr(obj, tuple(pre)) if pre else obj, post, val)
 
+def rgetattr(obj, attr, *args):
+    """https://stackoverflow.com/a/31174427/8783170"""
     def _getattr(obj, attr):
         return getattr(obj, attr, *args)
 
-    return functools.reduce(_getattr, [obj] + attr.split('.'))
+    return functools.reduce(_getattr, (obj,) + attr)

setup.cfg

@@ -9,7 +9,7 @@ name = pydantic_form
 author = theenglishway
 author_email = me@theenglishway.eu
 version = attr: pydantic_form.__version__
-description = Humanizen
+description = WTForms-based form with Pydantic backend
 long_description = file: README.rst, CHANGELOG.rst
 license = BSD 3-Clause License
 classifiers = 
@@ -30,11 +30,6 @@ exclude =
 	tests
 	tests.*
 
-[options.entry_points]
-flask.commands = 
-	seed = pydantic_form.commands:seed
-	clear = pydantic_form.commands:clear
-
 [bumpversion:file:pydantic_form/__init__.py]
 search = __version__ = '{current_version}'
 replace = __version__ = '{new_version}'

tests/conftest.py

@@ -5,48 +5,43 @@ from wtforms import Form, fields
 from werkzeug.datastructures import ImmutableMultiDict
 from pydantic_form import PydanticForm
 from pydantic import BaseModel, ValidationError
+from pydantic.types import StrictStr
 
 
-ScenarioClasses = namedtuple(
-    "ScenarioClasses",
-    ['wtf_form', 'pydantic_form', 'schema', 'data_factory', 'bad_data_factory', 'keys']
+
+DataFactories = namedtuple(
+    "DataFactories",
+    ['valid', 'bad', 'missing']
 )
 
-ScenarioInstances = namedtuple(
-    "ScenarioInstances",
-    [
-        'wtf', 'wtf_formdata',
-        'pydantic', 'pydantic_formdata',
-        'formdata', 'data',
-        'schema'
-    ]
+ExpectedErrors = namedtuple(
+    "ExpectedErrors",
+    ['valid', 'bad', 'missing']
 )
 
+Scenario = namedtuple(
+    "ScenarioClasses",
+    ['form', 'schema', 'keys', 'data_factory', 'errors']
+)
+
+
+@pytest.fixture
+def scenario(request):
+    return request.getfixturevalue(request.param)
 
-@pytest.fixture(scope="session")
-def instance_factory():
-    def _factory(scenario_classes, data):
-        formdata = ImmutableMultiDict(data)
-        try:
-            schema_instance = scenario_classes.schema(**data)
-        except ValidationError as e:
-            schema_instance = None
-        return ScenarioInstances(
-            scenario_classes.wtf_form(data=data),
-            scenario_classes.wtf_form(formdata=formdata),
-            scenario_classes.pydantic_form(data=data),
-            scenario_classes.pydantic_form(formdata=formdata),
-            formdata,
-            data,
-            schema_instance
-        )
-    return _factory
+
+class MissingDataFactory(factory.Factory):
+    class Meta:
+        model = dict
+
+
+# Simplest case
 
 simple_keys = [('integer',), ('string',)]
 
 class SimpleSchema(BaseModel):
     integer: int
-    string: str
+    string: StrictStr
 
 
 class SimpleWTForm(Form):
@@ -60,16 +55,6 @@ class SimpleForm(SimpleWTForm, PydanticForm):
     _schema = SimpleSchema
 
 
-@pytest.fixture(scope="session")
-def scenario_classes_simple():
-    return ScenarioClasses(
-        SimpleWTForm, SimpleForm,
-        SimpleSchema,
-        SimpleDataFactory, SimpleBadDataFactory,
-        simple_keys
-    )
-
-
 class SimpleDataFactory(factory.Factory):
     class Meta:
         model = dict
@@ -84,16 +69,35 @@ class SimpleBadDataFactory(factory.Factory):
     integer = factory.Faker('pystr')
     string = factory.Faker('pyint')
 
+simple_data_factories = DataFactories(
+    SimpleDataFactory,
+    SimpleBadDataFactory,
+    MissingDataFactory
+)
 
-@pytest.fixture
-def scenario_simple(instance_factory, scenario_classes_simple):
-    return instance_factory(scenario_classes_simple, SimpleDataFactory())
-
+simple_expected_errors = ExpectedErrors(
+    {},
+    {
+        ('integer',): 'type_error.integer',
+        ('string',): 'type_error.str'
+    },
+    {
+        ('integer',): 'value_error.missing',
+        ('string',): 'value_error.missing',
+    }
+)
 
 @pytest.fixture
-def scenario_simple_bad(instance_factory, scenario_classes_simple):
-    return instance_factory(scenario_classes_simple, SimpleBadDataFactory())
+def scenario_simple():
+    return Scenario(
+        SimpleForm,
+        SimpleSchema,
+        simple_keys,
+        simple_data_factories,
+        simple_expected_errors
+    )
 
+# Case with one level of nesting
 
 nested_keys = [('integer',), ('nested', 'integer'), ('nested', 'string')]
 
@@ -128,22 +132,102 @@ class NestedBadDataFactory(factory.Factory):
     integer = factory.Faker('pystr')
     nested = factory.SubFactory(SimpleBadDataFactory)
 
+nested_data_factories = DataFactories(
+    NestedDataFactory,
+    NestedBadDataFactory,
+    MissingDataFactory
+)
+
+nested_expected_errors = ExpectedErrors(
+    {},
+    {
+        ('integer',): 'type_error.integer',
+        ('nested', 'integer',): 'type_error.integer',
+        ('nested', 'string',): 'type_error.str'
+    },
+    {
+        ('integer',): 'value_error.missing',
+        ('nested',): 'value_error.missing',
+    }
+)
 
-@pytest.fixture(scope="session")
-def scenario_classes_nested():
-    return ScenarioClasses(
-        NestedWTForm, NestedForm,
+@pytest.fixture
+def scenario_nested():
+    return Scenario(
+        NestedForm,
         NestedSchema,
-        NestedDataFactory, NestedBadDataFactory,
-        nested_keys
+        nested_keys,
+        nested_data_factories,
+        nested_expected_errors
     )
 
 
-@pytest.fixture
-def scenario_nested(instance_factory, scenario_classes_nested):
-    return instance_factory(scenario_classes_nested, NestedDataFactory())
+# Case with two levels of nesting
+double_nested_keys = [
+    ('integer',),
+    ('double_nested', 'integer'),
+    ('double_nested', 'nested', 'integer'),
+    ('double_nested', 'nested', 'string')
+]
+
+class DoubleNestedSchema(BaseModel):
+    integer: int
+    double_nested: NestedSchema
 
 
+class DoubleNestedWTForm(Form):
+    _schema = DoubleNestedSchema
+
+    integer = fields.IntegerField()
+    double_nested = fields.FormField(form_class=NestedWTForm)
+
+
+class DoubleNestedForm(DoubleNestedWTForm, PydanticForm):
+    _schema = DoubleNestedSchema
+    double_nested = fields.FormField(form_class=NestedForm)
+
+
+class DoubleNestedDataFactory(factory.Factory):
+    class Meta:
+        model = dict
+
+    integer = factory.Faker('pyint')
+    double_nested = factory.SubFactory(NestedDataFactory)
+
+
+class DoubleNestedBadDataFactory(factory.Factory):
+    class Meta:
+        model = dict
+
+    integer = factory.Faker('pystr')
+    double_nested = factory.SubFactory(NestedBadDataFactory)
+
+double_nested_data_factories = DataFactories(
+    DoubleNestedDataFactory,
+    DoubleNestedBadDataFactory,
+    MissingDataFactory
+)
+
+double_nested_expected_errors = ExpectedErrors(
+    {},
+    {
+        ('integer',): 'type_error.integer',
+        ('double_nested', 'integer',): 'type_error.integer',
+        ('double_nested', 'nested', 'integer',): 'type_error.integer',
+        ('double_nested', 'nested', 'string',): 'type_error.str'
+    },
+    {
+        ('integer',): 'value_error.missing',
+        ('double_nested',): 'value_error.missing',
+    }
+)
+
 @pytest.fixture
-def scenario_nested_bad(instance_factory, scenario_classes_nested):
-    return instance_factory(scenario_classes_nested, NestedBadDataFactory())
+def scenario_double_nested():
+    return Scenario(
+        DoubleNestedForm,
+        DoubleNestedSchema,
+        double_nested_keys,
+        double_nested_data_factories,
+        double_nested_expected_errors
+    )

tests/test_iterators.py

@@ -0,0 +1,41 @@
+import pytest
+from pydantic_form.iterators import iter_form, iter_schema, iter_form_class
+
+
+@pytest.fixture
+def instance_factory(request):
+    def _factory(scenario, data):
+        instances = dict(
+            schema_class=scenario.schema,
+            schema=scenario.schema(**data),
+            form_class=scenario.wtf_form,
+            form=scenario.wtf_form()
+        )
+
+        return instances[request.param]
+
+    return _factory
+
+
+@pytest.mark.parametrize(
+    'scenario',
+    [
+        'scenario_simple',
+        'scenario_nested',
+        'scenario_double_nested'
+    ], indirect=True
+)
+def test_iterators(scenario):
+    data = scenario.data_factory.valid()
+    keys = scenario.keys
+
+    assert [k for k, _ in iter_form_class(scenario.form)] == keys
+
+    form = scenario.form()
+    assert [k for k, _ in iter_form_class(scenario.form)] == keys
+    assert [k for k, _ in iter_form(form)] == keys
+
+    assert [k for k, _ in iter_schema(scenario.schema)] == keys
+
+    schema_instance = scenario.schema(**data)
+    assert [k for k, _ in iter_schema(schema_instance)] == keys

tests/test_process.py

@@ -1,76 +1,117 @@
 from itertools import product
+from werkzeug.datastructures import ImmutableMultiDict
+from pydantic_form.translator import *
+from pydantic_form.utils import recursive_get
 import pytest
 
 
 @pytest.fixture
-def scenario(request):
-    return request.getfixturevalue(request.param)
+def data(request, scenario):
+    return getattr(scenario.data_factory, request.param)()
 
-FORMS_FORMDATA = [
-    'wtf_formdata', 'pydantic_formdata'
-]
-FORMS_DATA = [
-    'wtf',  'pydantic',
-]
-VALID_SCENARIOS = [
+SCENARIOS = [
     'scenario_simple',
-    'scenario_nested'
-]
-INVALID_SCENARIOS = [
-    'scenario_simple_bad',
-    'scenario_nested_bad',
+    'scenario_nested',
+    'scenario_double_nested'
 ]
 
-@pytest.mark.xfail
+@pytest.mark.parametrize(
+    'kwargs_factory',
+    [
+        lambda data: {'data': data},
+        lambda data: {'formdata': ImmutableMultiDict(data)},
+        lambda data: data,
+    ], ids=[
+        'data', 'formdata', 'kwargs'
+    ]
+)
+@pytest.mark.parametrize(
+    'data',
+    ['valid', 'bad'],
+    indirect=True
+)
 @pytest.mark.parametrize(
     'scenario',
-    VALID_SCENARIOS + INVALID_SCENARIOS,
+    SCENARIOS,
     indirect=True
 )
-def test_process(scenario):
-    assert scenario.wtf.data == scenario.pydantic.data
-    assert scenario.wtf_formdata.data == scenario.pydantic_formdata.data
+def test_process_data(scenario, data, kwargs_factory):
+    form = scenario.form()
+    kwargs = kwargs_factory(data)
+    form.process(**kwargs)
+    assert form.data == data
 
 
 @pytest.mark.parametrize(
-    'scenario, form_name',
-    product(VALID_SCENARIOS + INVALID_SCENARIOS, FORMS_FORMDATA + FORMS_DATA),
-    indirect=['scenario']
+    'data',
+    ['valid'],
+    indirect=True
 )
-def test_data(scenario, form_name):
-    form = getattr(scenario, form_name)
-    assert form.data == scenario.data
-
-
-@pytest.mark.xfail
 @pytest.mark.parametrize(
-    'scenario, form_name',
-    product(VALID_SCENARIOS, FORMS_FORMDATA + FORMS_DATA),
-    indirect=['scenario']
+    'scenario',
+    SCENARIOS,
+    indirect=True
 )
-def test_validate(scenario, form_name):
-    form = getattr(scenario, form_name)
+def test_validate_valid(scenario, data):
+    form = scenario.form(data=data)
+    assert form.validate()
+
+    form = scenario.form()
+    form.process(data=data)
     assert form.validate()
 
 
-@pytest.mark.xfail
 @pytest.mark.parametrize(
-    'scenario, form_name',
-    product(INVALID_SCENARIOS, FORMS_FORMDATA),
-    indirect=['scenario']
+    'data',
+    ['bad'],
+    indirect=True
+)
+@pytest.mark.parametrize(
+    'scenario',
+    SCENARIOS,
+    indirect=True
 )
-def test_errors_formdata(scenario, form_name):
-    form = getattr(scenario, form_name)
+def test_validate_bad(scenario, data):
+    form = scenario.form(data=data)
+    assert not form.validate()
+
+    form = scenario.form()
+    form.process(data=data)
     assert not form.validate()
-    assert form.errors
 
-@pytest.mark.xfail
+
+@pytest.mark.parametrize(
+    'data',
+    ['valid'],
+    indirect=True
+)
 @pytest.mark.parametrize(
-    'scenario, form_name',
-    product(INVALID_SCENARIOS, FORMS_DATA),
-    indirect=['scenario']
+    'scenario',
+    SCENARIOS,
+    indirect=True
 )
-def test_errors_data(scenario, form_name):
-    form = getattr(scenario, form_name)
-    assert form.validate()
-    assert form.errors == {}
+def test_errors_valid(scenario, data):
+    form = scenario.form(data=data)
+    assert form.errors == {}
+
+
+@pytest.mark.parametrize(
+    'data, errors_factory',
+    [
+        ('bad', lambda s: s.errors.bad),
+        ('missing', lambda s: s.errors.missing),
+    ],
+    indirect=['data']
+)
+@pytest.mark.parametrize(
+    'scenario',
+    SCENARIOS,
+    indirect=True
+)
+def test_errors_invalid(scenario, data, errors_factory):
+    form = scenario.form(data=data)
+
+    form.validate()
+    assert form.errors
+    for k, error in errors_factory(scenario).items():
+        assert recursive_get(form.errors, *k)[0]['type'] == error

tests/test_translator.py

@@ -1,59 +0,0 @@
-import pytest
-from pydantic_form.translator import SchemaInstanceToFormDataField, SchemaToForm
-
-
-@pytest.mark.parametrize('translator', [SchemaInstanceToFormDataField])
-@pytest.mark.parametrize(
-    'scenario_name',
-    ['scenario_classes_simple', 'scenario_classes_nested']
-)
-def test_data_translator(request, scenario_name, translator):
-    scenario = request.getfixturevalue(scenario_name)
-    data = scenario.data_factory()
-    schema = scenario.schema(**data)
-    form = scenario.wtf_form()
-    keys = scenario.keys
-
-    t = translator(schema, form)
-    for k in keys:
-        t(k)
-
-    assert schema.dict() == form.data
-
-@pytest.mark.skip
-@pytest.mark.parametrize('translator', [SchemaInstanceToFormDataField])
-@pytest.mark.parametrize(
-    'scenario_name',
-    ['scenario_classes_simple', 'scenario_classes_nested']
-)
-def test_error_translator_bad(request, scenario_name, translator):
-    scenario = request.getfixturevalue(scenario_name)
-    data = scenario.data_factory()
-    schema = scenario.schema(**data)
-    form = scenario.wtf_form()
-    keys = scenario.keys
-
-    t = translator(schema, form)
-    for k in keys:
-        t(k)
-
-    assert schema.dict() == form.data
-
-
-@pytest.mark.parametrize('translator', [SchemaToForm])
-@pytest.mark.parametrize('data_factory_name', ['data_factory', 'bad_data_factory'])
-@pytest.mark.parametrize(
-    'scenario_name',
-    ['scenario_classes_simple', 'scenario_classes_nested']
-)
-def test_translator(request, scenario_name, translator, data_factory_name):
-    scenario = request.getfixturevalue(scenario_name)
-    data = getattr(scenario, data_factory_name)()
-    schema = scenario.schema(**data)
-    form = scenario.wtf_form()
-    keys = scenario.keys
-
-    t = SchemaToForm(schema, form, keys)
-    t()
-
-    assert schema.dict() == form.data