1import functools 2 3from flex.datastructures import ValidationDict 4from flex.exceptions import ValidationError 5from flex.functional import chain_reduce_partial 6from flex.functional import attrgetter 7from flex.context_managers import ErrorDict 8from flex.http import ( 9 Request, 10) 11from flex.constants import ( 12 QUERY, 13 PATH, 14 HEADER, 15 BODY, 16) 17from flex.parameters import ( 18 filter_parameters, 19 merge_parameter_lists, 20 dereference_parameter_list, 21) 22from flex.validation.parameter import ( 23 validate_query_parameters, 24 construct_parameter_validators, 25) 26from flex.validation.header import ( 27 construct_header_validators, 28) 29from flex.validation.path import ( 30 generate_path_parameters_validator, 31) 32from flex.validation.common import ( 33 noop, 34 generate_value_processor, 35 generate_object_validator, 36 validate_content_type, 37) 38 39 40def validate_operation(request, validators, **kwargs): 41 with ErrorDict() as errors: 42 for key, validator in validators.items(): 43 try: 44 validator(request, **kwargs) 45 except ValidationError as err: 46 errors[key].add_error(err.detail) 47 48 49def validate_request_content_type(request, content_types, **kwargs): 50 assert isinstance(request, Request) 51 validate_content_type(request.content_type, content_types, **kwargs) 52 53 54def generate_request_content_type_validator(consumes, **kwargs): 55 validator = functools.partial( 56 validate_request_content_type, 57 content_types=consumes, 58 ) 59 return validator 60 61 62def generate_header_validator(headers, context, **kwargs): 63 """ 64 Generates a validation function that will validate a dictionary of headers. 65 """ 66 validators = ValidationDict() 67 for header_definition in headers: 68 header_processor = generate_value_processor( 69 context=context, 70 **header_definition 71 ) 72 header_validator = generate_object_validator( 73 field_validators=construct_header_validators(header_definition, context=context), 74 ) 75 validators.add_property_validator( 76 header_definition['name'], 77 chain_reduce_partial( 78 header_processor, 79 header_validator, 80 ), 81 ) 82 return generate_object_validator(field_validators=validators) 83 84 85def generate_form_data_validator(form_data_parameters, context, **kwargs): 86 pass 87 88 89def generate_request_body_validator(body_parameters, context, **kwargs): 90 if len(body_parameters) > 1: 91 raise ValueError("Too many body parameters. Should only be one") 92 elif not body_parameters: 93 return noop 94 body_validators = construct_parameter_validators( 95 body_parameters[0], context=context, 96 ) 97 return generate_object_validator(field_validators=body_validators) 98 99 100def generate_parameters_validator(api_path, path_definition, parameters, 101 context, **kwargs): 102 """ 103 Generates a validator function to validate. 104 105 - request.path against the path parameters. 106 - request.query against the query parameters. 107 - request.headers against the header parameters. 108 - TODO: request.body against the body parameters. 109 - TODO: request.formData against any form data. 110 """ 111 # TODO: figure out how to merge this with the same code in response 112 # validation. 113 validators = ValidationDict() 114 path_level_parameters = dereference_parameter_list( 115 path_definition.get('parameters', []), 116 context, 117 ) 118 operation_level_parameters = dereference_parameter_list( 119 parameters, 120 context, 121 ) 122 123 all_parameters = merge_parameter_lists( 124 path_level_parameters, 125 operation_level_parameters, 126 ) 127 128 # PATH 129 in_path_parameters = filter_parameters(all_parameters, in_=PATH) 130 validators.add_validator( 131 'path', 132 chain_reduce_partial( 133 attrgetter('path'), 134 generate_path_parameters_validator(api_path, in_path_parameters, context), 135 ), 136 ) 137 138 # QUERY 139 in_query_parameters = filter_parameters(all_parameters, in_=QUERY) 140 validators.add_validator( 141 'query', 142 chain_reduce_partial( 143 attrgetter('query_data'), 144 functools.partial( 145 validate_query_parameters, 146 query_parameters=in_query_parameters, 147 context=context, 148 ), 149 ), 150 ) 151 152 # HEADERS 153 in_header_parameters = filter_parameters(all_parameters, in_=HEADER) 154 validators.add_validator( 155 'headers', 156 chain_reduce_partial( 157 attrgetter('headers'), 158 generate_header_validator(in_header_parameters, context), 159 ), 160 ) 161 162 # FORM_DATA 163 # in_form_data_parameters = filter_parameters(all_parameters, in_=FORM_DATA) 164 # validators.add_validator( 165 # 'form_data', 166 # chain_reduce_partial( 167 # attrgetter('data'), 168 # generate_form_data_validator(in_form_data_parameters, context), 169 # ) 170 # ) 171 172 # REQUEST_BODY 173 in_request_body_parameters = filter_parameters(all_parameters, in_=BODY) 174 validators.add_validator( 175 'request_body', 176 chain_reduce_partial( 177 attrgetter('data'), 178 generate_request_body_validator(in_request_body_parameters, context), 179 ) 180 ) 181 182 return generate_object_validator(field_validators=validators) 183 184 185validator_mapping = { 186 'consumes': generate_request_content_type_validator, 187 'parameters': generate_parameters_validator, 188 'headers': generate_header_validator, 189} 190 191 192def construct_operation_validators(api_path, path_definition, operation_definition, context): 193 """ 194 - consumes (did the request conform to the content types this api consumes) 195 - produces (did the response conform to the content types this endpoint produces) 196 - parameters (did the parameters of this request validate) 197 TODO: move path parameter validation to here, because each operation 198 can override any of the path level parameters. 199 - schemes (was the request scheme correct) 200 - security: TODO since security isn't yet implemented. 201 """ 202 validators = {} 203 204 # sanity check 205 assert 'context' not in operation_definition 206 assert 'api_path' not in operation_definition 207 assert 'path_definition' not in operation_definition 208 209 for key in operation_definition.keys(): 210 if key not in validator_mapping: 211 # TODO: is this the right thing to do? 212 continue 213 validators[key] = validator_mapping[key]( 214 context=context, 215 api_path=api_path, 216 path_definition=path_definition, 217 **operation_definition 218 ) 219 220 # Global defaults 221 if 'consumes' in context and 'consumes' not in validators: 222 validators['consumes'] = validator_mapping['consumes'](**context) 223 if 'parameters' in path_definition and 'parameters' not in validators: 224 validators['parameters'] = validator_mapping['parameters']( 225 context=context, 226 api_path=api_path, 227 path_definition=path_definition, 228 parameters=path_definition['parameters'], 229 **operation_definition 230 ) 231 232 return validators 233