refactor: Modularize field creation in create_dynamic_schema

- Split field creation logic into helper functions for each attribute type.
- Improved readability, maintainability, and reusability of the code.
- No functional changes.
This commit is contained in:
Candifloss 2025-03-15 11:58:03 +05:30
parent 7714d64d6b
commit 2c8ad8a22a

View File

@ -1,10 +1,75 @@
from datetime import datetime
import re
from marshmallow import Schema, fields, ValidationError
from typing import Dict, Any, Optional, Type
from typing import Dict, Any, Optional, Type, Union
from config import item_attributes
from definitions.attributes import *
def create_text_field(attrib: textAttribute) -> fields.String:
"""Create a Marshmallow String field for textAttribute."""
def validate_allowed_chars(value: str, attrib=attrib) -> None:
if attrib.allowed_chars and not all(char in attrib.allowed_chars for char in value):
raise ValidationError(f"Invalid characters in {attrib.display_name}. Allowed characters are: {attrib.allowed_chars}")
return fields.String(
required=attrib.required,
validate=[
lambda x, attrib=attrib: len(x) <= attrib.max_length if attrib.max_length else True,
lambda x, attrib=attrib: len(x) >= attrib.min_length if attrib.min_length else True,
lambda x, attrib=attrib: re.match(attrib.regex, x) if attrib.regex else True,
validate_allowed_chars,
],
error_messages={
"required": f"{attrib.display_name} is required.",
"validator_failed": f"Invalid value for {attrib.display_name}.",
},
)
def create_numeric_field(attrib: Union[intAttribute, floatAttribute]) -> Union[fields.Integer, fields.Float]:
"""Create a Marshmallow Integer or Float field for intAttribute or floatAttribute."""
field_type = fields.Integer if isinstance(attrib, intAttribute) else fields.Float
return field_type(
required=attrib.required,
validate=[
lambda x, attrib=attrib: x >= attrib.min_val if attrib.min_val is not None else True,
lambda x, attrib=attrib: 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}.",
},
)
def create_date_field(attrib: dateAttribute) -> fields.Date:
"""Create a Marshmallow Date field for dateAttribute."""
return fields.Date(
required=attrib.required,
format="%Y-%m-%d",
validate=[
lambda x, attrib=attrib: x >= datetime.strptime(attrib.min_val, "%Y-%m-%d").date()
if attrib.min_val
else True,
lambda x, attrib=attrib: 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}.",
},
)
def create_select_field(attrib: selectAttribute) -> fields.String:
"""Create a Marshmallow String field for selectAttribute."""
return fields.String(
required=attrib.required,
validate=[lambda x, attrib=attrib: 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)}.",
},
)
def create_dynamic_schema() -> Type[Schema]:
"""
Dynamically creates a Marshmallow Schema based on the configuration in item_attributes.
@ -13,69 +78,13 @@ def create_dynamic_schema() -> Type[Schema]:
for attrib in item_attributes:
if isinstance(attrib, textAttribute):
# Allowed chars
def validate_allowed_chars(value: str, attrib=attrib) -> None:
if attrib.allowed_chars and not all(char in attrib.allowed_chars for char in value):
raise ValidationError(f"Invalid characters in {attrib.display_name}. Allowed characters are: {attrib.allowed_chars}")
schema_fields[attrib.attrib_name] = fields.String(
required=attrib.required,
validate=[
lambda x, attrib=attrib: len(x) <= attrib.max_length if attrib.max_length else True,
lambda x, attrib=attrib: len(x) >= attrib.min_length if attrib.min_length else True,
lambda x, attrib=attrib: re.match(attrib.regex, x) if attrib.regex else True,
validate_allowed_chars,
#validate compareto,
],
error_messages={
"required": f"{attrib.display_name} is required.",
"validator_failed": f"Invalid value for {attrib.display_name}.",
},
)
schema_fields[attrib.attrib_name] = create_text_field(attrib)
elif isinstance(attrib, (intAttribute, floatAttribute)):
# Determine the field type based on the attribute class
field_type = fields.Integer if isinstance(attrib, intAttribute) else fields.Float
schema_fields[attrib.attrib_name] = field_type(
required=attrib.required,
validate=[
lambda x, attrib=attrib: x >= attrib.min_val if attrib.min_val is not None else True,
lambda x, attrib=attrib: 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}.",
},
)
schema_fields[attrib.attrib_name] = create_numeric_field(attrib)
elif isinstance(attrib, dateAttribute):
schema_fields[attrib.attrib_name] = fields.Date(
required=attrib.required,
format="%Y-%m-%d",
validate=[
lambda x, attrib=attrib: x >= datetime.strptime(attrib.min_val, "%Y-%m-%d").date()
if attrib.min_val
else True,
lambda x, attrib=attrib: 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}.",
},
)
schema_fields[attrib.attrib_name] = create_date_field(attrib)
elif isinstance(attrib, selectAttribute):
schema_fields[attrib.attrib_name] = fields.String(
required=attrib.required,
validate=[lambda x, attrib=attrib: 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)}.",
},
)
schema_fields[attrib.attrib_name] = create_select_field(attrib)
# Dynamically create the schema class
DynamicSchema = type("DynamicSchema", (Schema,), schema_fields)