<s> from __future__ import unicode_literals <EOL> import decimal <EOL> import six <EOL> from collections import Hashable <EOL> from copy import deepcopy <EOL> from flask import current_app , request <EOL> from werkzeug . datastructures import MultiDict , FileStorage <EOL> from werkzeug import exceptions <EOL> from . errors import abort , SpecsError <EOL> from . marshalling import marshal <EOL> from . model import Model <EOL> class ParseResult ( dict ) : <EOL> '''<STR_LIT>''' <EOL> def __getattr__ ( self , name ) : <EOL> try : <EOL> return self [ name ] <EOL> except KeyError : <EOL> raise AttributeError ( name ) <EOL> def __setattr__ ( self , name , value ) : <EOL> self [ name ] = value <EOL> _friendly_location = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:args>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> LOCATIONS = { <EOL> '<STR_LIT:args>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:body>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> PY_TYPES = { <EOL> int : '<STR_LIT>' , <EOL> str : '<STR_LIT:string>' , <EOL> bool : '<STR_LIT>' , <EOL> None : '<STR_LIT>' <EOL> } <EOL> text_type = lambda x : six . text_type ( x ) <EOL> class Argument ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , name , default = None , dest = None , required = False , <EOL> ignore = False , type = text_type , location = ( '<STR_LIT>' , '<STR_LIT>' , ) , <EOL> choices = ( ) , action = '<STR_LIT:store>' , help = None , operators = ( '<STR_LIT:=>' , ) , <EOL> case_sensitive = True , store_missing = True , trim = False , <EOL> nullable = True ) : <EOL> self . name = name <EOL> self . default = default <EOL> self . dest = dest <EOL> self . required = required <EOL> self . ignore = ignore <EOL> self . location = location <EOL> self . type = type <EOL> self . choices = choices <EOL> self . action = action <EOL> self . help = help <EOL> self . case_sensitive = case_sensitive <EOL> self . operators = operators <EOL> self . store_missing = store_missing <EOL> self . trim = trim <EOL> self . nullable = nullable <EOL> def source ( self , request ) : <EOL> '''<STR_LIT>''' <EOL> if isinstance ( self . location , six . string_types ) : <EOL> value = getattr ( request , self . location , MultiDict ( ) ) <EOL> if callable ( value ) : <EOL> value = value ( ) <EOL> if value is not None : <EOL> return value <EOL> else : <EOL> values = MultiDict ( ) <EOL> for l in self . location : <EOL> value = getattr ( request , l , None ) <EOL> if callable ( value ) : <EOL> value = value ( ) <EOL> if value is not None : <EOL> values . update ( value ) <EOL> return values <EOL> return MultiDict ( ) <EOL> def convert ( self , value , op ) : <EOL> if value is None : <EOL> if not self . nullable : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return None <EOL> elif isinstance ( self . type , Model ) and isinstance ( value , dict ) : <EOL> return marshal ( value , self . type ) <EOL> elif isinstance ( value , FileStorage ) and self . type == FileStorage : <EOL> return value <EOL> try : <EOL> return self . type ( value , self . name , op ) <EOL> except TypeError : <EOL> try : <EOL> if self . type is decimal . Decimal : <EOL> return self . type ( str ( value ) , self . name ) <EOL> else : <EOL> return self . type ( value , self . name ) <EOL> except TypeError : <EOL> return self . type ( value ) <EOL> def handle_validation_error ( self , error , bundle_errors ) : <EOL> '''<STR_LIT>''' <EOL> error_str = six . text_type ( error ) <EOL> error_msg = self . help . format ( error_msg = error_str ) if self . help else error_str <EOL> errors = { self . name : error_msg } <EOL> if bundle_errors : <EOL> return error , errors <EOL> abort ( <NUM_LIT> , '<STR_LIT>' , errors = errors ) <EOL> def parse ( self , request , bundle_errors = False ) : <EOL> '''<STR_LIT>''' <EOL> bundle_errors = current_app . config . get ( '<STR_LIT>' , False ) or bundle_errors <EOL> source = self . source ( request ) <EOL> results = [ ] <EOL> _not_found = False <EOL> _found = True <EOL> for operator in self . operators : <EOL> name = self . name + operator . replace ( '<STR_LIT:=>' , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> if name in source : <EOL> if hasattr ( source , '<STR_LIT>' ) : <EOL> values = source . getlist ( name ) <EOL> else : <EOL> values = [ source . get ( name ) ] <EOL> for value in values : <EOL> if hasattr ( value , '<STR_LIT>' ) and self . trim : <EOL> value = value . strip ( ) <EOL> if hasattr ( value , '<STR_LIT>' ) and not self . case_sensitive : <EOL> value = value . lower ( ) <EOL> if hasattr ( self . choices , '<STR_LIT>' ) : <EOL> self . choices = [ choice . lower ( ) for choice in self . choices ] <EOL> try : <EOL> value = self . convert ( value , operator ) <EOL> except Exception as error : <EOL> if self . ignore : <EOL> continue <EOL> return self . handle_validation_error ( error , bundle_errors ) <EOL> if self . choices and value not in self . choices : <EOL> msg = '<STR_LIT>' . format ( value ) <EOL> if bundle_errors : <EOL> return self . handle_validation_error ( msg , bundle_errors ) <EOL> self . handle_validation_error ( msg , bundle_errors ) <EOL> if name in request . unparsed_arguments : <EOL> request . unparsed_arguments . pop ( name ) <EOL> results . append ( value ) <EOL> if not results and self . required : <EOL> if isinstance ( self . location , six . string_types ) : <EOL> location = _friendly_location . get ( self . location , self . location ) <EOL> else : <EOL> locations = [ _friendly_location . get ( loc , loc ) for loc in self . location ] <EOL> location = '<STR_LIT>' . join ( locations ) <EOL> error_msg = '<STR_LIT>' . format ( location ) <EOL> if bundle_errors : <EOL> return self . handle_validation_error ( error_msg , bundle_errors ) <EOL> self . handle_validation_error ( error_msg , bundle_errors ) <EOL> if not results : <EOL> if callable ( self . default ) : <EOL> return self . default ( ) , _not_found <EOL> else : <EOL> return self . default , _not_found <EOL> if self . action == '<STR_LIT>' : <EOL> return results , _found <EOL> if self . action == '<STR_LIT:store>' or len ( results ) == <NUM_LIT:1> : <EOL> return results [ <NUM_LIT:0> ] , _found <EOL> return results , _found <EOL> @ property <EOL> def __schema__ ( self ) : <EOL> if self . location == '<STR_LIT>' : <EOL> return <EOL> param = { <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : LOCATIONS . get ( self . location , '<STR_LIT>' ) <EOL> } <EOL> _handle_arg_type ( self , param ) <EOL> if self . required : <EOL> param [ '<STR_LIT>' ] = True <EOL> if self . help : <EOL> param [ '<STR_LIT:description>' ] = self . help <EOL> if self . default : <EOL> param [ '<STR_LIT:default>' ] = self . default <EOL> if self . action == '<STR_LIT>' : <EOL> param [ '<STR_LIT>' ] = { '<STR_LIT:type>' : param [ '<STR_LIT:type>' ] } <EOL> param [ '<STR_LIT:type>' ] = '<STR_LIT>' <EOL> param [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if self . choices : <EOL> param [ '<STR_LIT>' ] = self . choices <EOL> param [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return param <EOL> class RequestParser ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , argument_class = Argument , result_class = ParseResult , <EOL> trim = False , bundle_errors = False ) : <EOL> self . args = [ ] <EOL> self . argument_class = argument_class <EOL> self . result_class = result_class <EOL> self . trim = trim <EOL> self . bundle_errors = bundle_errors <EOL> def add_argument ( self , * args , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> if len ( args ) == <NUM_LIT:1> and isinstance ( args [ <NUM_LIT:0> ] , self . argument_class ) : <EOL> self . args . append ( args [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . args . append ( self . argument_class ( * args , ** kwargs ) ) <EOL> if self . trim and self . argument_class is Argument : <EOL> self . args [ - <NUM_LIT:1> ] . trim = kwargs . get ( '<STR_LIT>' , self . trim ) <EOL> return self <EOL> def parse_args ( self , req = None , strict = False ) : <EOL> '''<STR_LIT>''' <EOL> if req is None : <EOL> req = request <EOL> result = self . result_class ( ) <EOL> req . unparsed_arguments = dict ( self . argument_class ( '<STR_LIT>' ) . source ( req ) ) if strict else { } <EOL> errors = { } <EOL> for arg in self . args : <EOL> value , found = arg . parse ( req , self . bundle_errors ) <EOL> if isinstance ( value , ValueError ) : <EOL> errors . update ( found ) <EOL> found = None <EOL> if found or arg . store_missing : <EOL> result [ arg . dest or arg . name ] = value <EOL> if errors : <EOL> abort ( <NUM_LIT> , message = errors ) <EOL> if strict and req . unparsed_arguments : <EOL> arguments = '<STR_LIT:U+002CU+0020>' . join ( req . unparsed_arguments . keys ( ) ) <EOL> msg = '<STR_LIT>' . format ( arguments ) <EOL> raise exceptions . BadRequest ( msg ) <EOL> return result <EOL> def copy ( self ) : <EOL> '''<STR_LIT>''' <EOL> parser_copy = self . __class__ ( self . argument_class , self . result_class ) <EOL> parser_copy . args = deepcopy ( self . args ) <EOL> parser_copy . trim = self . trim <EOL> parser_copy . bundle_errors = self . bundle_errors <EOL> return parser_copy <EOL> def replace_argument ( self , name , * args , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> new_arg = self . argument_class ( name , * args , ** kwargs ) <EOL> for index , arg in enumerate ( self . args [ : ] ) : <EOL> if new_arg . name == arg . name : <EOL> del self . args [ index ] <EOL> self . args . append ( new_arg ) <EOL> break <EOL> return self <EOL> def remove_argument ( self , name ) : <EOL> '''<STR_LIT>''' <EOL> for index , arg in enumerate ( self . args [ : ] ) : <EOL> if name == arg . name : <EOL> del self . args [ index ] <EOL> break <EOL> return self <EOL> @ property <EOL> def __schema__ ( self ) : <EOL> params = [ ] <EOL> locations = set ( ) <EOL> for arg in self . args : <EOL> param = arg . __schema__ <EOL> if param : <EOL> params . append ( param ) <EOL> locations . add ( param [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT:body>' in locations and '<STR_LIT>' in locations : <EOL> raise SpecsError ( "<STR_LIT>" ) <EOL> return params <EOL> def _handle_arg_type ( arg , param ) : <EOL> if isinstance ( arg . type , Hashable ) and arg . type in PY_TYPES : <EOL> param [ '<STR_LIT:type>' ] = PY_TYPES [ arg . type ] <EOL> elif hasattr ( arg . type , '<STR_LIT>' ) : <EOL> param [ '<STR_LIT:type>' ] = arg . type . __apidoc__ [ '<STR_LIT:name>' ] <EOL> param [ '<STR_LIT>' ] = '<STR_LIT:body>' <EOL> elif hasattr ( arg . type , '<STR_LIT>' ) : <EOL> param . update ( arg . type . __schema__ ) <EOL> elif arg . location == '<STR_LIT>' : <EOL> param [ '<STR_LIT:type>' ] = '<STR_LIT:file>' <EOL> else : <EOL> param [ '<STR_LIT:type>' ] = '<STR_LIT:string>' </s>
