126 lines
4.9 KiB
Python
126 lines
4.9 KiB
Python
from datetime import datetime
|
|
import re
|
|
from marshmallow import Schema, fields, ValidationError, validates_schema
|
|
from typing import Dict, Any, Optional, Type
|
|
from config import item_attributes
|
|
from definitions.attribute import (
|
|
textAttribute,
|
|
intAttribute,
|
|
floatAttribute,
|
|
dateAttribute,
|
|
selectAttribute,
|
|
)
|
|
|
|
def create_dynamic_schema() -> Type[Schema]:
|
|
"""
|
|
Dynamically creates a marshmallow.Schema based on the configuration in item_attributes.
|
|
"""
|
|
class DynamicSchema(Schema):
|
|
class Meta:
|
|
strict = False # Ignore unknown fields
|
|
|
|
@validates_schema
|
|
def validate_required_fields(self, data: Dict[str, Any], **kwargs):
|
|
"""Ensure all required fields are present."""
|
|
for attrib in item_attributes:
|
|
if attrib.required and attrib.attrib_name not in data:
|
|
raise ValidationError(f"Missing required field: {attrib.display_name}.")
|
|
|
|
# Add fields to the schema based on item_attributes
|
|
for attrib in item_attributes:
|
|
print(f"Adding field: {attrib.attrib_name}") # Debugging
|
|
field = None
|
|
|
|
if isinstance(attrib, textAttribute):
|
|
field = fields.String(
|
|
required=attrib.required,
|
|
validate=[
|
|
lambda x: len(x) <= attrib.max_length if attrib.max_length else True,
|
|
lambda x: len(x) >= attrib.min_length if attrib.min_length else True,
|
|
lambda x: re.match(attrib.regex, x) if attrib.regex else True,
|
|
],
|
|
error_messages={
|
|
"required": f"{attrib.display_name} is required.",
|
|
"validator_failed": f"Invalid value for {attrib.display_name}.",
|
|
},
|
|
)
|
|
|
|
elif isinstance(attrib, intAttribute):
|
|
field = fields.Integer(
|
|
required=attrib.required,
|
|
validate=[
|
|
lambda x: x >= attrib.min_val if attrib.min_val is not None else True,
|
|
lambda x: x <= attrib.max_val if attrib.max_val is not None else True,
|
|
],
|
|
error_messages={
|
|
"required": f"{attrib.display_name} is required.",
|
|
"validator_failed": f"Invalid value for {attrib.display_name}.",
|
|
},
|
|
)
|
|
|
|
elif isinstance(attrib, floatAttribute):
|
|
field = fields.Float(
|
|
required=attrib.required,
|
|
validate=[
|
|
lambda x: x >= attrib.min_val if attrib.min_val is not None else True,
|
|
lambda x: x <= attrib.max_val if attrib.max_val is not None else True,
|
|
],
|
|
error_messages={
|
|
"required": f"{attrib.display_name} is required.",
|
|
"validator_failed": f"Invalid value for {attrib.display_name}.",
|
|
},
|
|
)
|
|
|
|
elif isinstance(attrib, dateAttribute):
|
|
field = fields.Date(
|
|
required=attrib.required,
|
|
format="%Y-%m-%d",
|
|
validate=[
|
|
lambda x: x >= datetime.strptime(attrib.min_val, "%Y-%m-%d").date()
|
|
if attrib.min_val
|
|
else True,
|
|
lambda x: x <= datetime.strptime(attrib.max_val, "%Y-%m-%d").date()
|
|
if attrib.max_val
|
|
else True,
|
|
],
|
|
error_messages={
|
|
"required": f"{attrib.display_name} is required.",
|
|
"validator_failed": f"Invalid value for {attrib.display_name}.",
|
|
},
|
|
)
|
|
|
|
elif isinstance(attrib, selectAttribute):
|
|
field = fields.String(
|
|
required=attrib.required,
|
|
validate=[lambda x: x in attrib.options],
|
|
error_messages={
|
|
"required": f"{attrib.display_name} is required.",
|
|
"validator_failed": f"Invalid value for {attrib.display_name}. Must be one of: {', '.join(attrib.options)}.",
|
|
},
|
|
)
|
|
|
|
if field:
|
|
#print(field)
|
|
setattr(DynamicSchema, attrib.attrib_name, field)
|
|
|
|
return DynamicSchema
|
|
|
|
def validate_values(form_data: Dict[str, Any]) -> Optional[str]:
|
|
"""
|
|
Validate form data against the configuration in item_attributes using marshmallow.
|
|
Returns an error message if invalid, otherwise None.
|
|
"""
|
|
DynamicSchema = create_dynamic_schema()
|
|
schema = DynamicSchema()
|
|
|
|
try:
|
|
schema.load(form_data) # Validate the data
|
|
print(form_data)
|
|
return None # No errors
|
|
except ValidationError as e:
|
|
# Format the error message for display
|
|
error_messages = []
|
|
for field, errors in e.messages.items():
|
|
for error in errors:
|
|
error_messages.append(f"{field}: {error}")
|
|
return "; ".join(error_messages) |