Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add MultiSource Parameter Type #49

Merged
merged 6 commits into from
Jul 11, 2024
Merged

Add MultiSource Parameter Type #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
abcb480
Update and reorganize README
smt5541 Jun 29, 2024
9e09847
Implement #43: Allow parameters to accept ingress from multiple possi…
smt5541 Jul 1, 2024
6577b2b
Update MultiSource to use *args for the expected delivery types, pass…
smt5541 Jul 1, 2024
c664ae4
Update README
smt5541 Jul 1, 2024
cae2a9a
Update README
smt5541 Jul 1, 2024
2fc5524
Fix merge conflicts in readme
Ge0rg3 Jul 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
206 changes: 133 additions & 73 deletions README.md
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 2 additions & 1 deletion flask_parameter_validation/parameter_types/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@
from .json import Json
from .query import Query
from .route import Route
from .multi_source import MultiSource

__all__ = [
"File", "Form", "Json", "Query", "Route"
"File", "Form", "Json", "Query", "Route", "MultiSource"
]
11 changes: 11 additions & 0 deletions flask_parameter_validation/parameter_types/multi_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
from typing import Type

from flask_parameter_validation.parameter_types.parameter import Parameter


class MultiSource(Parameter):
name = "multi_source"

def __init__(self, *sources: list[Type[Parameter]], **kwargs):
self.sources = [Source(**kwargs) for Source in sources]
super().__init__(**kwargs)
3 changes: 1 addition & 2 deletions flask_parameter_validation/parameter_types/parameter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
"""
import re
from datetime import date, datetime, time

import dateutil.parser as parser
import jsonschema
from jsonschema.exceptions import ValidationError as JSONSchemaValidationError
Expand Down Expand Up @@ -150,8 +151,6 @@ def validate(self, value):
if self.func is not None and not original_value_type_list:
self.func_helper(value)



return True

def convert(self, value, allowed_types):
Expand Down
203 changes: 109 additions & 94 deletions flask_parameter_validation/parameter_validation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from .exceptions import (InvalidParameterTypeError, MissingInputError,
ValidationError)
from .parameter_types import File, Form, Json, Query, Route
from .parameter_types.multi_source import MultiSource

fn_list = dict()

Expand Down Expand Up @@ -54,7 +55,7 @@ def nested_func_helper(**kwargs):
json_input = None
if request.headers.get("Content-Type") is not None:
if re.search(
"application/[^+]*[+]?(json);?", request.headers.get("Content-Type")
"application/[^+]*[+]?(json);?", request.headers.get("Content-Type")
):
try:
json_input = request.json
Expand Down Expand Up @@ -115,7 +116,7 @@ def nested_func(**kwargs):
return nested_func

def _to_dict_with_lists(
self, multi_dict: ImmutableMultiDict, expected_lists: list, split_strings: bool = False
self, multi_dict: ImmutableMultiDict, expected_lists: list, split_strings: bool = False
) -> dict:
dict_with_lists = {}
for key, values in multi_dict.lists():
Expand Down Expand Up @@ -155,108 +156,122 @@ def validate(self, expected_input, all_request_inputs):
original_expected_input_type = expected_input.annotation
original_expected_input_type_str = expected_input_type_str

# Validate that the expected delivery type is valid
if expected_delivery_type.__class__ not in all_request_inputs.keys():
raise InvalidParameterTypeError(expected_delivery_type)
# Expected delivery types can be a list if using MultiSource
expected_delivery_types = [expected_delivery_type]
if type(expected_delivery_type) is MultiSource:
expected_delivery_types = expected_delivery_type.sources

# Validate that user supplied input in expected delivery type (unless specified as Optional)
user_input = all_request_inputs[expected_delivery_type.__class__].get(
expected_name
)
if user_input is None:
# If default is given, set and continue
if expected_delivery_type.default is not None:
user_input = expected_delivery_type.default
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__
):
return user_input
for source_index, source in enumerate(expected_delivery_types):
# Validate that the expected delivery type is valid
if source.__class__ not in all_request_inputs.keys():
raise InvalidParameterTypeError(source)

# Validate that user supplied input in expected delivery type (unless specified as Optional)
user_input = all_request_inputs[source.__class__].get(
expected_name
)
if user_input is None:
# If default is given, set and continue
if source.default is not None:
user_input = source.default
else:
raise MissingInputError(
expected_name, expected_delivery_type.__class__
)
# 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__
and source_index == len(expected_delivery_types) - 1 # If MultiSource, only return None for last source
):
return user_input
else:
if len(expected_delivery_types) == 1:
raise MissingInputError(
expected_name, source.__class__
)
elif source_index != len(expected_delivery_types) - 1:
continue
else:
raise MissingInputError(
expected_name, source.__class__
)

# Skip validation if typing.Any is given
if expected_input_type_str.startswith("typing.Any"):
return user_input
# Skip validation if typing.Any is given
if expected_input_type_str.startswith("typing.Any"):
return user_input

# In python3.7+, typing.Optional is used instead of typing.Union[..., None]
if expected_input_type_str.startswith("typing.Optional"):
new_type = expected_input_type.__args__[0]
expected_input_type = new_type
expected_input_type_str = str(new_type)
# In python3.7+, typing.Optional is used instead of typing.Union[..., None]
if expected_input_type_str.startswith("typing.Optional"):
new_type = expected_input_type.__args__[0]
expected_input_type = new_type
expected_input_type_str = str(new_type)

# Prepare expected type checks for unions, lists and plain types
if expected_input_type_str.startswith("typing.Union"):
expected_input_types = expected_input_type.__args__
user_inputs = [user_input]
# If typing.List in union and user supplied valid list, convert remaining check only for list
for exp_type in expected_input_types:
if str(exp_type).startswith("typing.List"):
if type(user_input) is list:
# Only convert if validation passes
if hasattr(exp_type, "__args__"):
if all(type(inp) in exp_type.__args__ for inp in user_input):
expected_input_type = exp_type
expected_input_types = expected_input_type.__args__
expected_input_type_str = str(exp_type)
user_inputs = user_input
# If list, expand inner typing items. Otherwise, convert to list to match anyway.
elif expected_input_type_str.startswith("typing.List"):
expected_input_types = expected_input_type.__args__
if type(user_input) is list:
user_inputs = user_input
# Prepare expected type checks for unions, lists and plain types
if expected_input_type_str.startswith("typing.Union"):
expected_input_types = expected_input_type.__args__
user_inputs = [user_input]
# If typing.List in union and user supplied valid list, convert remaining check only for list
for exp_type in expected_input_types:
if str(exp_type).startswith("typing.List"):
if type(user_input) is list:
# Only convert if validation passes
if hasattr(exp_type, "__args__"):
if all(type(inp) in exp_type.__args__ for inp in user_input):
expected_input_type = exp_type
expected_input_types = expected_input_type.__args__
expected_input_type_str = str(exp_type)
user_inputs = user_input
# If list, expand inner typing items. Otherwise, convert to list to match anyway.
elif expected_input_type_str.startswith("typing.List"):
expected_input_types = expected_input_type.__args__
if type(user_input) is list:
user_inputs = user_input
else:
user_inputs = [user_input]
else:
user_inputs = [user_input]
else:
user_inputs = [user_input]
expected_input_types = [expected_input_type]
expected_input_types = [expected_input_type]

# Perform automatic type conversion for parameter types (i.e. "true" -> True)
for count, value in enumerate(user_inputs):
try:
user_inputs[count] = expected_delivery_type.convert(
value, expected_input_types
)
except ValueError as e:
raise ValidationError(str(e), expected_name, expected_input_type)
# Perform automatic type conversion for parameter types (i.e. "true" -> True)
for count, value in enumerate(user_inputs):
try:
user_inputs[count] = source.convert(
value, expected_input_types
)
except ValueError as e:
raise ValidationError(str(e), expected_name, expected_input_type)

# Validate that user type(s) match expected type(s)
validation_success = all(
type(inp) in expected_input_types for inp in user_inputs
)
# Validate that user type(s) match expected type(s)
validation_success = all(
type(inp) in expected_input_types for inp in user_inputs
)

# Validate that if lists are required, lists are given
if expected_input_type_str.startswith("typing.List"):
if type(user_input) is not list:
validation_success = False
# Validate that if lists are required, lists are given
if expected_input_type_str.startswith("typing.List"):
if type(user_input) is not list:
validation_success = False

# Error if types don't match
if not validation_success:
if hasattr(
original_expected_input_type, "__name__"
) and not original_expected_input_type_str.startswith("typing."):
type_name = original_expected_input_type.__name__
else:
type_name = original_expected_input_type_str
raise ValidationError(
f"must be type '{type_name}'",
expected_name,
original_expected_input_type,
)
# Error if types don't match
if not validation_success:
if hasattr(
original_expected_input_type, "__name__"
) and not original_expected_input_type_str.startswith("typing."):
type_name = original_expected_input_type.__name__
else:
type_name = original_expected_input_type_str
raise ValidationError(
f"must be type '{type_name}'",
expected_name,
original_expected_input_type,
)

# Validate parameter-specific requirements are met
try:
if type(user_input) is list:
expected_delivery_type.validate(user_input)
else:
expected_delivery_type.validate(user_inputs[0])
except ValueError as e:
raise ValidationError(str(e), expected_name, expected_input_type)
# Validate parameter-specific requirements are met
try:
if type(user_input) is list:
source.validate(user_input)
else:
source.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"):
return user_inputs
return user_inputs[0]
# Return input back to parent function
if expected_input_type_str.startswith("typing.List"):
return user_inputs
return user_inputs[0]
Loading
Loading