217 lines
9.2 KiB
Python
217 lines
9.2 KiB
Python
from flask_wtf import FlaskForm
|
|
from wtforms import IntegerField, FloatField, BooleanField, StringField, TextAreaField, validators, ValidationError
|
|
from flask import current_app
|
|
import json
|
|
|
|
from wtforms.fields.choices import SelectField
|
|
from wtforms.fields.datetime import DateField
|
|
|
|
|
|
class DynamicFormBase(FlaskForm):
|
|
def __init__(self, formdata=None, *args, **kwargs):
|
|
super(DynamicFormBase, self).__init__(*args, **kwargs)
|
|
# Maps collection names to lists of field names
|
|
self.dynamic_fields = {}
|
|
# Store formdata for later use
|
|
self.formdata = formdata
|
|
|
|
def _create_field_validators(self, field_def):
|
|
"""Create validators based on field definition"""
|
|
validators_list = []
|
|
|
|
# Required validator
|
|
if field_def.get('required', False):
|
|
validators_list.append(validators.InputRequired())
|
|
else:
|
|
validators_list.append(validators.Optional())
|
|
|
|
# Type-specific validators
|
|
field_type = field_def.get('type')
|
|
if field_type in ['integer', 'float']:
|
|
min_value = field_def.get('min_value')
|
|
max_value = field_def.get('max_value')
|
|
if min_value is not None or max_value is not None:
|
|
validators_list.append(
|
|
validators.NumberRange(
|
|
min=min_value if min_value is not None else -float('inf'),
|
|
max=max_value if max_value is not None else float('inf'),
|
|
message=f"Value must be between {min_value or '-∞'} and {max_value or '∞'}"
|
|
)
|
|
)
|
|
|
|
return validators_list
|
|
|
|
def add_dynamic_fields(self, collection_name, config, initial_data=None):
|
|
"""Add dynamic fields to the form based on the configuration."""
|
|
self.dynamic_fields[collection_name] = []
|
|
for field_name, field_def in config.items():
|
|
current_app.logger.debug(f"{field_name}: {field_def}")
|
|
# Prefix the field name with the collection name
|
|
full_field_name = f"{collection_name}_{field_name}"
|
|
label = field_def.get('name', field_name)
|
|
field_type = field_def.get('type')
|
|
description = field_def.get('description', '')
|
|
default = field_def.get('default')
|
|
|
|
# Determine standard validators
|
|
field_validators = self._create_field_validators(field_def)
|
|
|
|
# Handle special case for tagging_fields
|
|
if field_type == 'tagging_fields':
|
|
field_class = TextAreaField
|
|
field_validators.append(validate_tagging_fields)
|
|
extra_classes = 'json-editor'
|
|
field_kwargs = {}
|
|
elif field_type == 'enum':
|
|
field_class = SelectField
|
|
allowed_values = field_def.get('allowed_values', [])
|
|
choices = [(str(val), str(val)) for val in allowed_values]
|
|
extra_classes = ''
|
|
field_kwargs = {'choices': choices}
|
|
else:
|
|
extra_classes = ''
|
|
field_class = {
|
|
'integer': IntegerField,
|
|
'float': FloatField,
|
|
'boolean': BooleanField,
|
|
'string': StringField,
|
|
'date': DateField,
|
|
}.get(field_type, StringField)
|
|
field_kwargs = {}
|
|
|
|
# Prepare field data
|
|
field_data = None
|
|
if initial_data and field_name in initial_data:
|
|
field_data = initial_data[field_name]
|
|
if field_type == 'tagging_fields' and isinstance(field_data, dict):
|
|
try:
|
|
field_data = json.dumps(field_data, indent=2)
|
|
except (TypeError, ValueError) as e:
|
|
current_app.logger.error(f"Error converting initial data to JSON: {e}")
|
|
field_data = "{}"
|
|
elif default is not None:
|
|
field_data = default
|
|
|
|
# Create render_kw with classes and any other HTML attributes
|
|
render_kw = {'class': extra_classes} if extra_classes else {}
|
|
if description:
|
|
render_kw['title'] = description # For tooltip
|
|
render_kw['data-bs-toggle'] = 'tooltip'
|
|
render_kw['data-bs-placement'] = 'right'
|
|
|
|
# Create the field
|
|
field_kwargs.update({
|
|
'label': label,
|
|
'description': description,
|
|
'validators': field_validators,
|
|
'default': field_data,
|
|
'render_kw': render_kw
|
|
})
|
|
|
|
unbound_field = field_class(**field_kwargs)
|
|
|
|
# Bind the field to the form
|
|
bound_field = unbound_field.bind(form=self, name=full_field_name)
|
|
|
|
# Process the field with formdata
|
|
if self.formdata and full_field_name in self.formdata:
|
|
bound_field.process(self.formdata)
|
|
else:
|
|
bound_field.process(formdata=None, data=field_data) # Use prepared field_data
|
|
|
|
# Add the field to the form
|
|
setattr(self, full_field_name, bound_field)
|
|
self._fields[full_field_name] = bound_field
|
|
self.dynamic_fields[collection_name].append(full_field_name)
|
|
|
|
def get_static_fields(self):
|
|
"""Return a list of static field instances."""
|
|
# Get names of dynamic fields
|
|
dynamic_field_names = set()
|
|
for field_list in self.dynamic_fields.values():
|
|
dynamic_field_names.update(field_list)
|
|
|
|
# Return all fields that are not dynamic
|
|
return [field for name, field in self._fields.items() if name not in dynamic_field_names]
|
|
|
|
def get_dynamic_fields(self):
|
|
"""Return a dictionary of dynamic fields per collection."""
|
|
result = {}
|
|
for collection_name, field_names in self.dynamic_fields.items():
|
|
result[collection_name] = [getattr(self, name) for name in field_names]
|
|
return result
|
|
|
|
def get_dynamic_data(self, collection_name):
|
|
"""Retrieve the data from dynamic fields of a specific collection."""
|
|
data = {}
|
|
current_app.logger.debug(f"{collection_name} in {self.dynamic_fields}?")
|
|
if collection_name not in self.dynamic_fields:
|
|
return data
|
|
prefix_length = len(collection_name) + 1 # +1 for the underscore
|
|
for full_field_name in self.dynamic_fields[collection_name]:
|
|
current_app.logger.debug(f"{full_field_name}: {full_field_name}")
|
|
original_field_name = full_field_name[prefix_length:]
|
|
current_app.logger.debug(f"{original_field_name}: {original_field_name}")
|
|
field = getattr(self, full_field_name)
|
|
current_app.logger.debug(f"{field}: {field}")
|
|
# Parse JSON for tagging_fields type
|
|
if isinstance(field, TextAreaField) and field.data:
|
|
try:
|
|
data[original_field_name] = json.loads(field.data)
|
|
except json.JSONDecodeError:
|
|
# Validation should catch this, but just in case
|
|
data[original_field_name] = field.data
|
|
else:
|
|
data[original_field_name] = field.data
|
|
return data
|
|
|
|
|
|
def validate_tagging_fields(form, field):
|
|
"""Validate the tagging fields structure"""
|
|
if not field.data:
|
|
return
|
|
|
|
try:
|
|
# Parse JSON data
|
|
fields_data = json.loads(field.data)
|
|
|
|
# Validate it's a dictionary
|
|
if not isinstance(fields_data, dict):
|
|
raise ValidationError("Tagging fields must be a dictionary")
|
|
|
|
# Validate each field definition
|
|
for field_name, field_def in fields_data.items():
|
|
if not isinstance(field_def, dict):
|
|
raise ValidationError(f"Field definition for {field_name} must be a dictionary")
|
|
|
|
# Check required properties
|
|
if 'type' not in field_def:
|
|
raise ValidationError(f"Field {field_name} missing required 'type' property")
|
|
|
|
# Validate type
|
|
if field_def['type'] not in ['string', 'integer', 'float', 'date', 'enum']:
|
|
raise ValidationError(f"Field {field_name} has invalid type: {field_def['type']}")
|
|
|
|
# Validate enum fields have allowed_values
|
|
if field_def['type'] == 'enum':
|
|
if 'allowed_values' not in field_def:
|
|
raise ValidationError(f"Enum field {field_name} missing required 'allowed_values' list")
|
|
if not isinstance(field_def['allowed_values'], list):
|
|
raise ValidationError(f"Field {field_name} allowed_values must be a list")
|
|
|
|
# Validate numeric fields
|
|
if field_def['type'] in ['integer', 'float']:
|
|
if 'min_value' in field_def and 'max_value' in field_def:
|
|
min_val = float(field_def['min_value'])
|
|
max_val = float(field_def['max_value'])
|
|
if min_val >= max_val:
|
|
raise ValidationError(f"Field {field_name} min_value must be less than max_value")
|
|
|
|
except json.JSONDecodeError:
|
|
raise ValidationError("Invalid JSON format")
|
|
except (TypeError, ValueError) as e:
|
|
raise ValidationError(f"Invalid field definition: {str(e)}")
|
|
|
|
|
|
|