<s> from __future__ import unicode_literals <EOL> from collections import namedtuple <EOL> from contextlib import contextmanager <EOL> import functools <EOL> import logging <EOL> import re <EOL> import sys <EOL> import bravado_core <EOL> from bravado_core . exception import SwaggerMappingError <EOL> from bravado_core . formatter import SwaggerFormat <EOL> from bravado_core . request import RequestLike , unmarshal_request <EOL> from bravado_core . response import get_response_spec , OutgoingResponse <EOL> from pyramid . interfaces import IRoutesMapper <EOL> import jsonschema . exceptions <EOL> import simplejson <EOL> from pyramid_swagger . exceptions import RequestValidationError <EOL> from pyramid_swagger . exceptions import ResponseValidationError <EOL> from pyramid_swagger . model import PathNotMatchedError <EOL> log = logging . getLogger ( __name__ ) <EOL> SWAGGER_12 = '<STR_LIT>' <EOL> SWAGGER_20 = '<STR_LIT>' <EOL> DEFAULT_SWAGGER_VERSIONS = [ SWAGGER_20 ] <EOL> SUPPORTED_SWAGGER_VERSIONS = [ SWAGGER_12 , SWAGGER_20 ] <EOL> DEFAULT_EXCLUDED_PATHS = [ <EOL> r'<STR_LIT>' , <EOL> r'<STR_LIT>' , <EOL> r'<STR_LIT>' , <EOL> ] <EOL> class Settings ( namedtuple ( <EOL> '<STR_LIT>' , <EOL> [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> ) ) : <EOL> """<STR_LIT>""" <EOL> @ contextmanager <EOL> def noop_context ( request , response = None ) : <EOL> yield <EOL> def _get_validation_context ( registry ) : <EOL> validation_context_path = registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> if validation_context_path : <EOL> m = re . match ( <EOL> '<STR_LIT>' , <EOL> validation_context_path , <EOL> ) <EOL> module_path = m . group ( '<STR_LIT>' ) <EOL> contextmanager_name = m . group ( '<STR_LIT>' ) <EOL> return getattr ( <EOL> __import__ ( module_path , fromlist = contextmanager_name ) , <EOL> contextmanager_name , <EOL> ) <EOL> else : <EOL> return noop_context <EOL> def get_swagger_objects ( settings , route_info , registry ) : <EOL> """<STR_LIT>""" <EOL> enabled_swagger_versions = get_swagger_versions ( registry . settings ) <EOL> schema12 = registry . settings [ '<STR_LIT>' ] <EOL> schema20 = registry . settings [ '<STR_LIT>' ] <EOL> fallback_to_swagger12_route = ( <EOL> SWAGGER_20 in enabled_swagger_versions and <EOL> SWAGGER_12 in enabled_swagger_versions and <EOL> settings . prefer_20_routes and <EOL> route_info . get ( '<STR_LIT>' ) and <EOL> route_info [ '<STR_LIT>' ] . name not in settings . prefer_20_routes <EOL> ) <EOL> if fallback_to_swagger12_route : <EOL> return settings . swagger12_handler , schema12 <EOL> if SWAGGER_20 in enabled_swagger_versions : <EOL> return settings . swagger20_handler , schema20 <EOL> if SWAGGER_12 in enabled_swagger_versions : <EOL> return settings . swagger12_handler , schema12 <EOL> def validation_tween_factory ( handler , registry ) : <EOL> """<STR_LIT>""" <EOL> settings = load_settings ( registry ) <EOL> route_mapper = registry . queryUtility ( IRoutesMapper ) <EOL> def validator_tween ( request ) : <EOL> route_info = route_mapper ( request ) <EOL> swagger_handler , spec = get_swagger_objects ( settings , route_info , <EOL> registry ) <EOL> if should_exclude_request ( settings , request , route_info ) : <EOL> return handler ( request ) <EOL> validation_context = _get_validation_context ( registry ) <EOL> try : <EOL> op_or_validators_map = swagger_handler . op_for_request ( <EOL> request , route_info = route_info , spec = spec ) <EOL> except PathNotMatchedError as exc : <EOL> if settings . validate_path : <EOL> with validation_context ( request ) : <EOL> raise RequestValidationError ( str ( exc ) ) <EOL> else : <EOL> return handler ( request ) <EOL> if settings . validate_request : <EOL> request_data = swagger_handler . handle_request ( <EOL> PyramidSwaggerRequest ( request , route_info ) , <EOL> op_or_validators_map , <EOL> validation_context = validation_context ) <EOL> def swagger_data ( _ ) : <EOL> return request_data <EOL> request . set_property ( swagger_data ) <EOL> response = handler ( request ) <EOL> if settings . validate_response : <EOL> with validation_context ( request , response = response ) : <EOL> swagger_handler . handle_response ( response , op_or_validators_map ) <EOL> return response <EOL> return validator_tween <EOL> class PyramidSwaggerRequest ( RequestLike ) : <EOL> """<STR_LIT>""" <EOL> FORM_TYPES = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> def __init__ ( self , request , route_info ) : <EOL> """<STR_LIT>""" <EOL> self . request = request <EOL> self . route_info = route_info <EOL> @ property <EOL> def query ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . request . GET . mixed ( ) <EOL> @ property <EOL> def path ( self ) : <EOL> return self . route_info . get ( '<STR_LIT>' ) or { } <EOL> @ property <EOL> def headers ( self ) : <EOL> return self . request . headers <EOL> @ property <EOL> def form ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . request . content_type in self . FORM_TYPES : <EOL> return self . request . POST . mixed ( ) <EOL> return { } <EOL> @ property <EOL> def body ( self ) : <EOL> return self . json ( ) <EOL> @ property <EOL> def files ( self ) : <EOL> result = { } <EOL> for k , v in self . request . params . mixed ( ) . items ( ) : <EOL> if hasattr ( v , '<STR_LIT:file>' ) : <EOL> result [ k ] = v . file <EOL> return result <EOL> def json ( self , ** kwargs ) : <EOL> return getattr ( self . request , '<STR_LIT>' , { } ) <EOL> class PyramidSwaggerResponse ( OutgoingResponse ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , response ) : <EOL> """<STR_LIT>""" <EOL> self . response = response <EOL> @ property <EOL> def content_type ( self ) : <EOL> return self . response . content_type <EOL> @ property <EOL> def text ( self ) : <EOL> return self . response . text <EOL> def json ( self , ** kwargs ) : <EOL> return getattr ( self . response , '<STR_LIT>' , { } ) <EOL> def handle_request ( request , validator_map , validation_context , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> request_data = { } <EOL> validation_pairs = [ ] <EOL> for validator , values in [ <EOL> ( validator_map . query , request . query ) , <EOL> ( validator_map . path , request . path ) , <EOL> ( validator_map . form , request . form ) , <EOL> ( validator_map . headers , request . headers ) , <EOL> ] : <EOL> values = cast_params ( validator . schema , values ) <EOL> validation_pairs . append ( ( validator , values ) ) <EOL> request_data . update ( values ) <EOL> if validator_map . body . schema : <EOL> param_name = validator_map . body . schema [ '<STR_LIT:name>' ] <EOL> validation_pairs . append ( ( validator_map . body , request . body ) ) <EOL> request_data [ param_name ] = request . body <EOL> with validation_context ( request ) : <EOL> validate_request ( validation_pairs ) <EOL> return request_data <EOL> def load_settings ( registry ) : <EOL> return Settings ( <EOL> swagger12_handler = build_swagger12_handler ( <EOL> registry . settings . get ( '<STR_LIT>' ) ) , <EOL> swagger20_handler = build_swagger20_handler ( ) , <EOL> validate_request = registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> True <EOL> ) , <EOL> validate_response = registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> True <EOL> ) , <EOL> validate_path = registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> True <EOL> ) , <EOL> exclude_paths = get_exclude_paths ( registry ) , <EOL> exclude_routes = set ( registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> ) or [ ] ) , <EOL> prefer_20_routes = set ( registry . settings . get ( <EOL> '<STR_LIT>' , ) or [ ] ) , <EOL> ) <EOL> SwaggerHandler = namedtuple ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def build_swagger20_handler ( ) : <EOL> return SwaggerHandler ( <EOL> op_for_request = get_op_for_request , <EOL> handle_request = swaggerize_request , <EOL> handle_response = swaggerize_response , <EOL> ) <EOL> def build_swagger12_handler ( schema ) : <EOL> """<STR_LIT>""" <EOL> if schema : <EOL> return SwaggerHandler ( <EOL> op_for_request = schema . validators_for_request , <EOL> handle_request = handle_request , <EOL> handle_response = validate_response , <EOL> ) <EOL> def get_exclude_paths ( registry ) : <EOL> """<STR_LIT>""" <EOL> regexes = registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> registry . settings . get ( <EOL> '<STR_LIT>' , <EOL> DEFAULT_EXCLUDED_PATHS <EOL> ) <EOL> ) <EOL> if not isinstance ( regexes , list ) and not isinstance ( regexes , tuple ) : <EOL> regexes = [ regexes ] <EOL> return [ re . compile ( r ) for r in regexes ] <EOL> def should_exclude_request ( settings , request , route_info ) : <EOL> disable_all_validation = not any ( ( <EOL> settings . validate_request , <EOL> settings . validate_response , <EOL> settings . validate_path <EOL> ) ) <EOL> return ( <EOL> disable_all_validation or <EOL> should_exclude_path ( settings . exclude_paths , request . path ) or <EOL> should_exclude_route ( settings . exclude_routes , route_info ) <EOL> ) <EOL> def should_exclude_path ( exclude_path_regexes , path ) : <EOL> return any ( r . match ( path ) for r in exclude_path_regexes ) <EOL> def should_exclude_route ( excluded_routes , route_info ) : <EOL> return ( <EOL> route_info . get ( '<STR_LIT>' ) and <EOL> route_info [ '<STR_LIT>' ] . name in excluded_routes <EOL> ) <EOL> def validation_error ( exc_class ) : <EOL> def decorator ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def _validate ( * args , ** kwargs ) : <EOL> try : <EOL> return f ( * args , ** kwargs ) <EOL> except ( jsonschema . exceptions . ValidationError , <EOL> SwaggerMappingError ) as exc : <EOL> e = exc_class ( str ( exc ) ) <EOL> e . _traceback = sys . exc_info ( ) [ <NUM_LIT:2> ] <EOL> raise e <EOL> return _validate <EOL> return decorator <EOL> CAST_TYPE_TO_FUNC = { <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT:float>' : float , <EOL> '<STR_LIT>' : float , <EOL> '<STR_LIT>' : bool , <EOL> } <EOL> def cast_request_param ( param_type , param_name , param_value ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return CAST_TYPE_TO_FUNC . get ( param_type , lambda x : x ) ( param_value ) <EOL> except ValueError : <EOL> log . warn ( "<STR_LIT>" , <EOL> param_name , param_value , param_type ) <EOL> return param_value <EOL> @ validation_error ( RequestValidationError ) <EOL> def validate_request ( validation_pairs ) : <EOL> for validator , values in validation_pairs : <EOL> validator . validate ( values ) <EOL> def cast_params ( schema , values ) : <EOL> if not schema : <EOL> return { } <EOL> def get_type ( param_name ) : <EOL> return schema [ '<STR_LIT>' ] . get ( param_name , { } ) . get ( '<STR_LIT:type>' ) <EOL> return dict ( <EOL> ( k , cast_request_param ( get_type ( k ) , k , v ) ) <EOL> for k , v in values . items ( ) <EOL> ) <EOL> @ validation_error ( ResponseValidationError ) <EOL> def validate_response ( response , validator_map ) : <EOL> """<STR_LIT>""" <EOL> validator = validator_map . response <EOL> returns_nothing = validator . schema . get ( '<STR_LIT:type>' ) == '<STR_LIT>' <EOL> body_empty = response . body in ( None , b'<STR_LIT>' , b'<STR_LIT:{}>' , b'<STR_LIT:null>' ) <EOL> if returns_nothing and body_empty : <EOL> return <EOL> if not <NUM_LIT:200> <= response . status_code <= <NUM_LIT> : <EOL> return <EOL> validator . validate ( prepare_body ( response ) ) <EOL> def prepare_body ( response ) : <EOL> if response . content_type is None or response . charset is None : <EOL> raise ResponseValidationError ( <EOL> '<STR_LIT>' <EOL> ) <EOL> if '<STR_LIT:application/json>' in response . content_type : <EOL> return simplejson . loads ( response . text ) <EOL> else : <EOL> return response . text <EOL> @ validation_error ( RequestValidationError ) <EOL> def swaggerize_request ( request , op , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> validation_context = kwargs [ '<STR_LIT>' ] <EOL> with validation_context ( request ) : <EOL> request_data = unmarshal_request ( request , op ) <EOL> return request_data <EOL> @ validation_error ( ResponseValidationError ) <EOL> def swaggerize_response ( response , op ) : <EOL> """<STR_LIT>""" <EOL> response_spec = get_response_spec ( response . status_int , op ) <EOL> bravado_core . response . validate_response ( <EOL> response_spec , op , PyramidSwaggerResponse ( response ) ) <EOL> def get_op_for_request ( request , route_info , spec ) : <EOL> """<STR_LIT>""" <EOL> route = route_info [ '<STR_LIT>' ] <EOL> if hasattr ( route , '<STR_LIT:path>' ) : <EOL> route_path = route . path <EOL> if route_path [ <NUM_LIT:0> ] != '<STR_LIT:/>' : <EOL> route_path = '<STR_LIT:/>' + route_path <EOL> op = spec . get_op_for_request ( request . method , route_path ) <EOL> if op is not None : <EOL> return op <EOL> raise PathNotMatchedError ( <EOL> "<STR_LIT>" <EOL> . format ( request . method , request . url ) ) <EOL> def get_swagger_versions ( settings ) : <EOL> """<STR_LIT>""" <EOL> swagger_versions = settings . get ( <EOL> '<STR_LIT>' , DEFAULT_SWAGGER_VERSIONS ) <EOL> if len ( swagger_versions ) == <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> for swagger_version in swagger_versions : <EOL> if swagger_version not in SUPPORTED_SWAGGER_VERSIONS : <EOL> raise ValueError ( '<STR_LIT>' <EOL> . format ( swagger_version ) ) <EOL> return swagger_versions </s>
