in azure_functions_worker/functions.py [0:0]
def validate_function_params(params: dict, bound_params: dict,
annotations: dict, func_name: str):
if set(params) - set(bound_params):
raise FunctionLoadError(
func_name,
'the following parameters are declared in Python but '
f'not in function.json: {set(params) - set(bound_params)!r}')
if set(bound_params) - set(params):
raise FunctionLoadError(
func_name,
f'the following parameters are declared in function.json but '
f'not in Python: {set(bound_params) - set(params)!r}')
input_types: typing.Dict[str, ParamTypeInfo] = {}
output_types: typing.Dict[str, ParamTypeInfo] = {}
fx_deferred_bindings_enabled = False
for param in params.values():
binding = bound_params[param.name]
param_has_anno = param.name in annotations
param_anno = annotations.get(param.name)
# Check if deferred bindings is enabled
fx_deferred_bindings_enabled, is_deferred_binding = (
bindings_utils.check_deferred_bindings_enabled(
param_anno,
fx_deferred_bindings_enabled))
if param_has_anno:
if typing_inspect.is_generic_type(param_anno):
param_anno_origin = typing_inspect.get_origin(param_anno)
if param_anno_origin is not None:
is_param_out = (
isinstance(param_anno_origin, type)
and param_anno_origin.__name__ == 'Out'
)
else:
is_param_out = (
isinstance(param_anno, type)
and param_anno.__name__ == 'Out'
)
else:
is_param_out = (
isinstance(param_anno, type)
and param_anno.__name__ == 'Out'
)
else:
is_param_out = False
is_binding_out = binding.direction == protos.BindingInfo.out
if is_param_out:
param_anno_args = typing_inspect.get_args(param_anno)
if len(param_anno_args) != 1:
raise FunctionLoadError(
func_name,
f'binding {param.name} has invalid Out annotation '
f'{param_anno!r}')
param_py_type = param_anno_args[0]
# typing_inspect.get_args() returns a flat list,
# so if the annotation was func.Out[typing.List[foo]],
# we need to reconstruct it.
if (isinstance(param_py_type, tuple)
and typing_inspect.is_generic_type(param_py_type[0])):
param_py_type = operator.getitem(
param_py_type[0], *param_py_type[1:])
else:
param_py_type = param_anno
if (param_has_anno and not isinstance(param_py_type, type)
and not typing_inspect.is_generic_type(param_py_type)):
raise FunctionLoadError(
func_name,
f'binding {param.name} has invalid non-type annotation '
f'{param_anno!r}')
if is_binding_out and param_has_anno and not is_param_out:
raise FunctionLoadError(
func_name,
f'binding {param.name} is declared to have the "out" '
'direction, but its annotation in Python is not '
'a subclass of azure.functions.Out')
if not is_binding_out and is_param_out:
raise FunctionLoadError(
func_name,
f'binding {param.name} is declared to have the "in" '
'direction in function.json, but its annotation '
'is azure.functions.Out in Python')
if param_has_anno and param_py_type in (str, bytes) and (
not bindings_utils.has_implicit_output(binding.type)):
param_bind_type = 'generic'
else:
param_bind_type = binding.type
if param_has_anno:
if is_param_out:
checks_out = bindings_utils.check_output_type_annotation(
param_bind_type, param_py_type)
else:
checks_out = bindings_utils.check_input_type_annotation(
param_bind_type, param_py_type, is_deferred_binding)
if not checks_out:
if binding.data_type is not protos.BindingInfo.undefined:
raise FunctionLoadError(
func_name,
f'{param.name!r} binding type "{binding.type}" '
f'and dataType "{binding.data_type}" in '
f'function.json do not match the corresponding '
f'function parameter\'s Python type '
f'annotation "{param_py_type.__name__}"')
else:
raise FunctionLoadError(
func_name,
f'type of {param.name} binding in function.json '
f'"{binding.type}" does not match its Python '
f'annotation "{param_py_type.__name__}"')
param_type_info = ParamTypeInfo(param_bind_type,
param_py_type,
is_deferred_binding)
if is_binding_out:
output_types[param.name] = param_type_info
else:
input_types[param.name] = param_type_info
return input_types, output_types, fx_deferred_bindings_enabled