WIP: Dictionary (JSON) Validation from Json and Query

Author: smt5541

flask_parameter_validation/parameter_types/parameter.py

@@ -2,10 +2,13 @@
     Base Parameter class.
     Should only be used as child class for other params.
 """
+import json
 import re
 from datetime import date, datetime, time
 
 import dateutil.parser as parser
+import jsonschema
+from jsonschema.exceptions import ValidationError as JSONSchemaValidationError
 
 
 class Parameter:
@@ -27,6 +30,7 @@ def __init__(
         datetime_format=None,  # str: datetime format string (https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes),
         comment=None,  # str: comment for autogenerated documentation
         alias=None,  # str: alias for parameter name
+        json_schema=None,  # dict: JSON Schema to check received dicts or lists against
     ):
         self.default = default
         self.min_list_length = min_list_length
@@ -42,9 +46,11 @@ def __init__(
         self.datetime_format = datetime_format
         self.comment = comment
         self.alias = alias
+        self.json_schema = json_schema
 
     # Validator
     def validate(self, value):
+        print(f"Parameter#validate received {type(value)}")
         if type(value) is list:
             values = value
             # Min list len
@@ -59,6 +65,21 @@ def validate(self, value):
                     raise ValueError(
                         f"must have have a maximum of {self.max_list_length} items."
                     )
+            if self.json_schema is not None:
+                try:
+                    jsonschema.validate(value, self.json_schema)
+                    print("JSON Schema Passed")
+                except JSONSchemaValidationError as e:
+                    raise ValueError(f"failed JSON Schema validation: {e.args[0]}")
+        elif type(value) is dict:
+            print("dict validate path")
+            if self.json_schema is not None:
+                try:
+                    jsonschema.validate(value, self.json_schema)
+                except JSONSchemaValidationError as e:
+                    print("Failed JSON Schema")
+                    raise ValueError(f"failed JSON Schema validation: {e.args[0]}")
+            values = [value]
         else:
             values = [value]
 
@@ -133,6 +154,7 @@ def validate(self, value):
 
     def convert(self, value, allowed_types):
         """Some parameter types require manual type conversion (see Query)"""
+        print(f"Parameter#convert received {type(value)}")
         # Datetime conversion
         if None in allowed_types and value is None:
             return value
@@ -160,4 +182,5 @@ def convert(self, value, allowed_types):
                 return date.fromisoformat(str(value))
             except ValueError:
                 raise ValueError("date format does not match ISO 8601")
+        print(f"Parameter#convert returned {type(value)}")
         return value

flask_parameter_validation/parameter_types/query.py

@@ -2,6 +2,8 @@
     Query Parameters
     - i.e. sent in GET requests, /?username=myname
 """
+import json
+
 from .parameter import Parameter
 
 
@@ -13,6 +15,7 @@ def __init__(self, default=None, **kwargs):
 
     def convert(self, value, allowed_types):
         """Convert query parameters to corresponding types."""
+        print(f"Query#convert received {type(value)}")
         if type(value) is str:
             # int conversion
             if int in allowed_types:
@@ -32,5 +35,11 @@ def convert(self, value, allowed_types):
                     value = True
                 elif value.lower() == "false":
                     value = False
-
+            # dict conversion
+            if dict in allowed_types:
+                try:
+                    value = json.loads(value)
+                except ValueError:
+                    raise ValueError(f"invalid JSON")
+        print(f"Query#convert returned {type(value)}")
         return super().convert(value, allowed_types)

flask_parameter_validation/parameter_validation.py

@@ -1,6 +1,7 @@
 import asyncio
 import functools
 import inspect
+import json
 import re
 from inspect import signature
 
@@ -46,7 +47,7 @@ def __call__(self, f):
         async def nested_func(**kwargs):
             # Step 1 - Get expected input details as dict
             expected_inputs = signature(f).parameters
-            
+
             # Step 2 - Validate JSON inputs
             json_input = None
             if request.headers.get("Content-Type") is not None:
@@ -61,7 +62,8 @@ async def nested_func(**kwargs):
             # Step 3 - Extract list of parameters expected to be lists (otherwise all values are converted to lists)
             expected_list_params = []
             for name, param in expected_inputs.items():
-                if str(param.annotation).startswith("typing.List") or str(param.annotation).startswith("typing.Optional[typing.List"):
+                if str(param.annotation).startswith("typing.List") or str(param.annotation).startswith(
+                        "typing.Optional[typing.List"):
                     expected_list_params.append(param.default.alias or name)
 
             # Step 4 - Convert request inputs to dicts
@@ -152,7 +154,7 @@ def validate(self, expected_input, all_request_inputs):
             else:
                 # Optionals are Unions with a NoneType, so we should check if None is part of Union __args__ (if exist)
                 if (
-                    hasattr(expected_input_type, "__args__") and type(None) in expected_input_type.__args__
+                        hasattr(expected_input_type, "__args__") and type(None) in expected_input_type.__args__
                 ):
                     return user_input
                 else:
@@ -202,6 +204,9 @@ def validate(self, expected_input, all_request_inputs):
                 user_inputs[count] = expected_delivery_type.convert(
                     value, expected_input_types
                 )
+                for t in expected_input_types:
+                    print(f"Expected type: {t}")
+                print(f"Appended value of type {type(user_inputs[count])} to user_inputs")
             except ValueError as e:
                 raise ValidationError(str(e), expected_name, expected_input_type)
 
@@ -215,6 +220,8 @@ def validate(self, expected_input, all_request_inputs):
             if type(user_input) is not list:
                 validation_success = False
 
+        print(f"Validation Success? {validation_success}")
+
         # Error if types don't match
         if not validation_success:
             if hasattr(
@@ -229,11 +236,19 @@ def validate(self, expected_input, all_request_inputs):
                 original_expected_input_type,
             )
 
-        # Validate parameter-specific requirements are met
-        try:
-            expected_delivery_type.validate(user_input)
-        except ValueError as e:
-            raise ValidationError(str(e), expected_name, expected_input_type)
+        print(json.dumps(user_input))
+
+        if expected_input_type_str.startswith("typing.List"):
+            # Validate parameter-specific requirements are met
+            try:
+                expected_delivery_type.validate(user_input)
+            except ValueError as e:
+                raise ValidationError(str(e), expected_name, expected_input_type)
+        else:
+            try:
+                expected_delivery_type.validate(user_inputs[0])
+            except ValueError as e:
+                raise ValidationError(str(e), expected_name, expected_input_type)
 
         # Return input back to parent function
         if expected_input_type_str.startswith("typing.List"):

setup.py

@@ -30,6 +30,7 @@
         'Flask',
         'flask[async]',
         'python-dateutil',
+        'jsonschema',
     ],
     classifiers=[
         'Environment :: Web Environment',