This is a gist for data validation for cloud firestore nosql db.
The function takes 2 parameter a datastructure and data, the datastructure that contains all the validation fields such as minlength, default, required etc, it is as show in the Example datastructure. The data is the data passed by user that will be validated
def clean_data(data_structure, data, update=False):
"""
validates the data and returns errors if any and the validated data
"""
errors = {}
validated_data = {}
def addError(error, key):
if key in errors:
if isinstance(error, list):
errors[key].extends(error)
else:
errors[key].append(error)
else:
errors[key] = [error]
# print("Data: ", data, type(data))
for key, rules in data_structure.items():
if (update and rules.get('oneoff') and key in data): # if the data is oneoff only add the value once
addError(f"'{key}' cannot be updated", key)
if rules.get('readonly', False) == True:
# if the readonly is true then the user value must be discarded and default value must be used
if check_type(rules.get('default'), Callable):
if (not update or not rules.get('oneoff')): # if the data is oneoff only add the value once
data[key] = rules.get('default')() # continue with the validation
else:
raise TypeError("Invalid default value, must be a callable function")
if key not in data and rules.get('required', False) == True and rules.get('readonly', False) == False:
# if required is true and the key doesn't exist then raise error
addError(f"'{key}' is required but not provided.", key)
continue
if (rules.get('required') == False or update) and key not in data:
continue # don't continue with the validation if the key doesn't exist, and required is False
else:
value = data[key] # if it exists continue with the validation
if 'type' in rules and not check_type(value, rules['type']):
addError(f"'{key}' should be of type {rules['type']}.", key)
if 'minlength' in rules and len(value) < rules['minlength']:
addError(f"'{key}' is the below the minimum length of {rules['minlength']}.", key)
if 'maxlength' in rules and len(value) > rules['maxlength']:
addError(f"'{key}' exceeds the maximum length of {rules['maxlength']}.", key)
if 'validators' in rules:
for validator in rules['validators']:
validated = validator(value)
if validated != True:
addError(f"{validated}", key)
validated_data[key] = value
if 'inner' in rules:
if isinstance(rules['inner'], dict):
nested_validated_data, nested_errors = clean_data(rules['inner'], value)
# print("validated: ", nested_validated_data, nested_errors)
if nested_validated_data:
validated_data[key] = nested_validated_data
if nested_errors:
# raise ValidationException()
addError([f"'{key}' has {error}" for error in nested_errors], key)
elif check_type(rules['inner'], List[Dict]):
_rule = rules['inner']
if rules['repeat'] and len(value) > 1:
_rule.extend(_rule*len(value))
for idx, x in enumerate(_rule):
# print("validated: ", x, value, end="\n\n")
nested_validated_data, nested_errors = clean_data(x, value[idx])
if nested_validated_data:
validated_data[key] = nested_validated_data
if nested_errors:
# raise ValidationException()
addError(nested_errors, key)
return validated_data, errorsThe datastructure that is used to validate the user data.
class FieldValue(TypedDict):
type: str
minlength: int
maxlength: int
required: bool
validators: List
inner: Union[Dict, List]
default: Callable
readonly: bool
oneoff: bool # The value can't be chaned if this is True
DatastructureType = Dict[str, FieldValue]
CREATE_BLOG: DatastructureType = {
'title': {'type': str, 'maxlength': 150, 'required': True},
'blog': {'type': str, 'required': True},
'tags': {'type': List[str], 'maxlength': 3, 'required': True, 'validators': [tag_validator]},
'social': {
'inner':{
'reddit': {'type': str, 'maxlength':30, 'required': False, 'validators': [name_validator]},
'stackoverflow': {'type': str, 'maxlength':30, 'required': False, 'validators': [name_validator]},
'twitter': {'type': str, 'maxlength':30, 'required': False, 'validators': [name_validator]},
'mastodon': {'type': str, 'maxlength':30, 'required': False, 'validators': [name_validator]},
'discord': {'type': str, 'maxlength':30, 'required': False, 'validators': [name_validator]},
}
},
'additional_links': {
'type': List[Dict],
'minlength': 0,
'maxlength': 2,
'required': False,
'repeat': True, # set this if the inner has to repeat
'inner': [
{
'link_name': {'type': str, 'maxlength': 20, 'required': True},
'link_url': {'type': str, 'required': True, 'validators': [url_validator]}
}
]
},
'sponsor': {
'inner': {
'buymeacoffee': {'type': str, 'maxlength': 25, 'required': False, 'validators': [name_validator]},
'patreon': {'type': str, 'maxlength': 25, 'required': False, 'validators': [name_validator]},
}
},
'datetime': {'type': datetime, 'default': timezone.now, 'required': True, 'readonly': True, 'oneoff': True},
'user': {'type': int, 'required': True}
}