<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import copy <EOL> import os <EOL> import re <EOL> import traceback <EOL> from collections import defaultdict <EOL> import six <EOL> from pants . base . deprecated import deprecated_conditional , validate_removal_semver , warn_or_error <EOL> from pants . option . arg_splitter import GLOBAL_SCOPE , GLOBAL_SCOPE_CONFIG_SECTION <EOL> from pants . option . custom_types import ( ListValueComponent , dict_option , file_option , list_option , <EOL> target_list_option , target_option ) <EOL> from pants . option . errors import ( BooleanOptionNameWithNo , FrozenRegistration , ImplicitValIsNone , <EOL> InvalidKwarg , InvalidMemberType , MemberTypeNotAllowed , <EOL> NoOptionNames , OptionAlreadyRegistered , OptionNameDash , <EOL> OptionNameDoubleDash , ParseError , RecursiveSubsystemOption , <EOL> Shadowing ) <EOL> from pants . option . option_util import is_list_option <EOL> from pants . option . ranked_value import RankedValue <EOL> from pants . option . scope import ScopeInfo <EOL> class Parser ( object ) : <EOL> """<STR_LIT>""" <EOL> class BooleanConversionError ( ParseError ) : <EOL> """<STR_LIT>""" <EOL> class FromfileError ( ParseError ) : <EOL> """<STR_LIT>""" <EOL> @ staticmethod <EOL> def _ensure_bool ( s ) : <EOL> if isinstance ( s , six . string_types ) : <EOL> if s . lower ( ) == '<STR_LIT:true>' : <EOL> return True <EOL> elif s . lower ( ) == '<STR_LIT:false>' : <EOL> return False <EOL> else : <EOL> raise Parser . BooleanConversionError ( '<STR_LIT>' . format ( s ) ) <EOL> if s is True : <EOL> return True <EOL> elif s is False : <EOL> return False <EOL> else : <EOL> raise Parser . BooleanConversionError ( '<STR_LIT>' . format ( s ) ) <EOL> @ classmethod <EOL> def _invert ( cls , s ) : <EOL> if s is None : <EOL> return None <EOL> b = cls . _ensure_bool ( s ) <EOL> return not b <EOL> def __init__ ( self , env , config , scope_info , parent_parser , option_tracker ) : <EOL> """<STR_LIT>""" <EOL> self . _env = env <EOL> self . _config = config <EOL> self . _scope_info = scope_info <EOL> self . _scope = self . _scope_info . scope <EOL> self . _option_tracker = option_tracker <EOL> self . _frozen = False <EOL> self . _known_args = set ( ) <EOL> self . _option_registrations = [ ] <EOL> self . _parent_parser = parent_parser <EOL> self . _child_parsers = [ ] <EOL> if self . _parent_parser : <EOL> self . _parent_parser . _register_child_parser ( self ) <EOL> @ property <EOL> def scope ( self ) : <EOL> return self . _scope <EOL> def walk ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> callback ( self ) <EOL> for child in self . _child_parsers : <EOL> child . walk ( callback ) <EOL> def _create_flag_value_map ( self , flags ) : <EOL> """<STR_LIT>""" <EOL> flag_value_map = defaultdict ( list ) <EOL> for flag in flags : <EOL> key , has_equals_sign , flag_val = flag . partition ( '<STR_LIT:=>' ) <EOL> if not has_equals_sign : <EOL> if not flag . startswith ( '<STR_LIT>' ) : <EOL> key = flag [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> flag_val = flag [ <NUM_LIT:2> : ] <EOL> if not flag_val : <EOL> flag_val = None <EOL> flag_value_map [ key ] . append ( flag_val ) <EOL> return flag_value_map <EOL> def parse_args ( self , flags , namespace ) : <EOL> """<STR_LIT>""" <EOL> flag_value_map = self . _create_flag_value_map ( flags ) <EOL> for args , kwargs in self . _unnormalized_option_registrations_iter ( ) : <EOL> self . _validate ( args , kwargs ) <EOL> dest = kwargs . get ( '<STR_LIT>' ) or self . _select_dest ( args ) <EOL> def consume_flag ( flag ) : <EOL> self . _check_deprecated ( dest , kwargs ) <EOL> del flag_value_map [ flag ] <EOL> implicit_value = kwargs . get ( '<STR_LIT>' ) <EOL> if implicit_value is None and kwargs . get ( '<STR_LIT:type>' ) == bool : <EOL> implicit_value = True <EOL> flag_vals = [ ] <EOL> def add_flag_val ( v ) : <EOL> if v is None : <EOL> if implicit_value is None : <EOL> raise ParseError ( '<STR_LIT>' . format ( <EOL> arg , self . _scope_str ( ) ) ) <EOL> else : <EOL> flag_vals . append ( implicit_value ) <EOL> else : <EOL> flag_vals . append ( v ) <EOL> for arg in args : <EOL> if kwargs . get ( '<STR_LIT:type>' ) == bool : <EOL> inverse_arg = self . _inverse_arg ( arg ) <EOL> if inverse_arg in flag_value_map : <EOL> flag_value_map [ arg ] = [ self . _invert ( v ) for v in flag_value_map [ inverse_arg ] ] <EOL> implicit_value = self . _invert ( implicit_value ) <EOL> del flag_value_map [ inverse_arg ] <EOL> if arg in flag_value_map : <EOL> for v in flag_value_map [ arg ] : <EOL> add_flag_val ( v ) <EOL> consume_flag ( arg ) <EOL> try : <EOL> val = self . _compute_value ( dest , kwargs , flag_vals ) <EOL> except ParseError as e : <EOL> raise type ( e ) ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( '<STR_LIT:U+002CU+0020>' . join ( args ) , self . _scope_str ( ) , traceback . format_exc ( ) ) <EOL> ) <EOL> setattr ( namespace , dest , val ) <EOL> if flag_value_map : <EOL> raise ParseError ( '<STR_LIT>' . format ( <EOL> self . _scope_str ( ) , '<STR_LIT:U+002CU+0020>' . join ( flag_value_map . keys ( ) ) ) ) <EOL> return namespace <EOL> def option_registrations_iter ( self ) : <EOL> """<STR_LIT>""" <EOL> def normalize_kwargs ( args , orig_kwargs ) : <EOL> nkwargs = copy . copy ( orig_kwargs ) <EOL> dest = nkwargs . get ( '<STR_LIT>' ) or self . _select_dest ( args ) <EOL> nkwargs [ '<STR_LIT>' ] = dest <EOL> if not ( '<STR_LIT:default>' in nkwargs and isinstance ( nkwargs [ '<STR_LIT:default>' ] , RankedValue ) ) : <EOL> nkwargs [ '<STR_LIT:default>' ] = self . _compute_value ( dest , nkwargs , [ ] ) <EOL> return nkwargs <EOL> if self . _parent_parser : <EOL> for args , kwargs in self . _parent_parser . _recursive_option_registration_args ( ) : <EOL> yield args , normalize_kwargs ( args , kwargs ) <EOL> for args , kwargs in self . _option_registrations : <EOL> normalized_kwargs = normalize_kwargs ( args , kwargs ) <EOL> if '<STR_LIT>' in normalized_kwargs : <EOL> normalized_kwargs [ '<STR_LIT>' ] = True <EOL> yield args , normalized_kwargs <EOL> def _unnormalized_option_registrations_iter ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _parent_parser : <EOL> for args , kwargs in self . _parent_parser . _recursive_option_registration_args ( ) : <EOL> yield args , kwargs <EOL> for args , kwargs in self . _option_registrations : <EOL> if '<STR_LIT>' in kwargs and self . _scope_info . category == ScopeInfo . SUBSYSTEM : <EOL> raise RecursiveSubsystemOption ( self . scope , args [ <NUM_LIT:0> ] ) <EOL> yield args , kwargs <EOL> def _recursive_option_registration_args ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _parent_parser : <EOL> for args , kwargs in self . _parent_parser . _recursive_option_registration_args ( ) : <EOL> yield args , kwargs <EOL> for args , kwargs in self . _option_registrations : <EOL> if self . _scope_info . category == ScopeInfo . SUBSYSTEM or '<STR_LIT>' in kwargs : <EOL> yield args , kwargs <EOL> def register ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . _frozen : <EOL> raise FrozenRegistration ( self . scope , args [ <NUM_LIT:0> ] ) <EOL> ancestor = self . _parent_parser <EOL> while ancestor : <EOL> ancestor . _freeze ( ) <EOL> ancestor = ancestor . _parent_parser <EOL> def check_deprecated_types ( kwarg_name ) : <EOL> t = kwargs . get ( kwarg_name ) <EOL> if t == list_option : <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , <EOL> self . scope ) ) <EOL> elif t == dict_option : <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , <EOL> self . scope ) ) <EOL> check_deprecated_types ( '<STR_LIT:type>' ) <EOL> check_deprecated_types ( '<STR_LIT>' ) <EOL> action = kwargs . get ( '<STR_LIT:action>' ) <EOL> if action == '<STR_LIT>' : <EOL> if '<STR_LIT:type>' in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = kwargs [ '<STR_LIT:type>' ] <EOL> kwargs [ '<STR_LIT:type>' ] = list <EOL> del kwargs [ '<STR_LIT:action>' ] <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , "<STR_LIT>" , <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , self . scope ) ) <EOL> if kwargs . get ( '<STR_LIT:type>' ) == target_list_option : <EOL> kwargs [ '<STR_LIT:type>' ] = list <EOL> kwargs [ '<STR_LIT>' ] = target_option <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , self . scope ) ) <EOL> if action == '<STR_LIT:store_true>' : <EOL> kwargs [ '<STR_LIT:type>' ] = bool <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> del kwargs [ '<STR_LIT:action>' ] <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , "<STR_LIT>" , <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , self . scope ) ) <EOL> elif action == '<STR_LIT>' : <EOL> kwargs [ '<STR_LIT:type>' ] = bool <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> del kwargs [ '<STR_LIT:action>' ] <EOL> deprecated_conditional ( lambda : True , '<STR_LIT>' , "<STR_LIT>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( args [ <NUM_LIT:0> ] , self . scope ) ) <EOL> if kwargs . get ( '<STR_LIT:type>' ) == bool and kwargs . get ( '<STR_LIT:default>' ) is None : <EOL> kwargs [ '<STR_LIT:default>' ] = not self . _ensure_bool ( kwargs . get ( '<STR_LIT>' , True ) ) <EOL> self . _option_registrations . append ( ( args , kwargs ) ) <EOL> if self . _parent_parser : <EOL> for arg in args : <EOL> existing_scope = self . _parent_parser . _existing_scope ( arg ) <EOL> if existing_scope is not None : <EOL> raise Shadowing ( self . scope , arg , outer_scope = self . _scope_str ( existing_scope ) ) <EOL> for arg in args : <EOL> if arg in self . _known_args : <EOL> raise OptionAlreadyRegistered ( self . scope , arg ) <EOL> self . _known_args . update ( args ) <EOL> def _check_deprecated ( self , dest , kwargs ) : <EOL> """<STR_LIT>""" <EOL> removal_version = kwargs . get ( '<STR_LIT>' , None ) <EOL> if removal_version is not None : <EOL> warn_or_error ( removal_version , <EOL> "<STR_LIT>" . format ( dest , self . _scope_str ( ) ) , <EOL> kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT> ) <EOL> _allowed_registration_kwargs = { <EOL> '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:default>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> } <EOL> _allowed_member_types = { <EOL> str , int , float , dict , dict_option , file_option , target_option <EOL> } <EOL> def _validate ( self , args , kwargs ) : <EOL> """<STR_LIT>""" <EOL> def error ( exception_type , arg_name = None , ** msg_kwargs ) : <EOL> if arg_name is None : <EOL> arg_name = args [ <NUM_LIT:0> ] if args else '<STR_LIT>' <EOL> raise exception_type ( self . scope , arg_name , ** msg_kwargs ) <EOL> if not args : <EOL> error ( NoOptionNames ) <EOL> for arg in args : <EOL> if not arg . startswith ( '<STR_LIT:->' ) : <EOL> error ( OptionNameDash , arg_name = arg ) <EOL> if not arg . startswith ( '<STR_LIT>' ) and len ( arg ) > <NUM_LIT:2> : <EOL> error ( OptionNameDoubleDash , arg_name = arg ) <EOL> if '<STR_LIT>' in kwargs and kwargs [ '<STR_LIT>' ] is None : <EOL> error ( ImplicitValIsNone ) <EOL> if '<STR_LIT>' in kwargs and kwargs . get ( '<STR_LIT:type>' , str ) not in [ list , list_option ] : <EOL> error ( MemberTypeNotAllowed , type_ = kwargs . get ( '<STR_LIT:type>' , str ) . __name__ ) <EOL> if kwargs . get ( '<STR_LIT>' , str ) not in self . _allowed_member_types : <EOL> error ( InvalidMemberType , member_type = kwargs . get ( '<STR_LIT>' , str ) . __name__ ) <EOL> for kwarg in kwargs : <EOL> if kwarg not in self . _allowed_registration_kwargs : <EOL> error ( InvalidKwarg , kwarg = kwarg ) <EOL> removal_version = kwargs . get ( '<STR_LIT>' ) <EOL> if removal_version is not None : <EOL> validate_removal_semver ( removal_version ) <EOL> def _existing_scope ( self , arg ) : <EOL> if arg in self . _known_args : <EOL> return self . _scope <EOL> elif self . _parent_parser : <EOL> return self . _parent_parser . _existing_scope ( arg ) <EOL> else : <EOL> return None <EOL> _ENV_SANITIZER_RE = re . compile ( r'<STR_LIT>' ) <EOL> def _select_dest ( self , args ) : <EOL> """<STR_LIT>""" <EOL> arg = next ( ( a for a in args if a . startswith ( '<STR_LIT>' ) ) , args [ <NUM_LIT:0> ] ) <EOL> return arg . lstrip ( '<STR_LIT:->' ) . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) <EOL> @ staticmethod <EOL> def _wrap_type ( t ) : <EOL> if t == list : <EOL> return list_option <EOL> elif t == dict : <EOL> return dict_option <EOL> else : <EOL> return t <EOL> def _compute_value ( self , dest , kwargs , flag_val_strs ) : <EOL> """<STR_LIT>""" <EOL> def to_value_type ( val_str ) : <EOL> if val_str is None : <EOL> return None <EOL> elif kwargs . get ( '<STR_LIT:type>' ) == bool : <EOL> return self . _ensure_bool ( val_str ) <EOL> else : <EOL> return self . _wrap_type ( kwargs . get ( '<STR_LIT:type>' , str ) ) ( val_str ) <EOL> def expand ( val_str ) : <EOL> if kwargs . get ( '<STR_LIT>' , False ) and val_str and val_str . startswith ( '<STR_LIT:@>' ) : <EOL> if val_str . startswith ( '<STR_LIT>' ) : <EOL> return val_str [ <NUM_LIT:1> : ] <EOL> else : <EOL> fromfile = val_str [ <NUM_LIT:1> : ] <EOL> try : <EOL> with open ( fromfile ) as fp : <EOL> return fp . read ( ) . strip ( ) <EOL> except IOError as e : <EOL> raise self . FromfileError ( '<STR_LIT>' . format ( <EOL> dest , self . _scope_str ( ) , fromfile , e ) ) <EOL> else : <EOL> return val_str <EOL> config_details = None <EOL> config_section = GLOBAL_SCOPE_CONFIG_SECTION if self . _scope == GLOBAL_SCOPE else self . _scope <EOL> config_val_str = expand ( self . _config . get ( config_section , dest , default = None ) ) <EOL> config_source_file = self . _config . get_source_for_option ( config_section , dest ) <EOL> if config_source_file is not None : <EOL> config_source_file = os . path . relpath ( config_source_file ) <EOL> config_details = '<STR_LIT>' . format ( config_source_file ) <EOL> udest = dest . upper ( ) <EOL> if self . _scope == GLOBAL_SCOPE : <EOL> env_vars = [ '<STR_LIT>' . format ( udest ) , '<STR_LIT>' . format ( udest ) ] <EOL> if udest . startswith ( '<STR_LIT>' ) : <EOL> env_vars . append ( udest ) <EOL> else : <EOL> sanitized_env_var_scope = self . _ENV_SANITIZER_RE . sub ( '<STR_LIT:_>' , self . _scope . upper ( ) ) <EOL> env_vars = [ '<STR_LIT>' . format ( sanitized_env_var_scope , udest ) ] <EOL> env_val_str = None <EOL> env_details = None <EOL> if self . _env : <EOL> for env_var in env_vars : <EOL> if env_var in self . _env : <EOL> env_val_str = expand ( self . _env . get ( env_var ) ) <EOL> env_details = '<STR_LIT>' . format ( env_var ) <EOL> break <EOL> flag_vals = [ to_value_type ( expand ( x ) ) for x in flag_val_strs ] <EOL> if is_list_option ( kwargs ) : <EOL> flag_val = ListValueComponent . merge ( flag_vals ) if flag_vals else None <EOL> elif len ( flag_vals ) > <NUM_LIT:1> : <EOL> raise ParseError ( '<STR_LIT>' . format ( <EOL> dest , self . _scope_str ( ) ) ) <EOL> elif len ( flag_vals ) == <NUM_LIT:1> : <EOL> flag_val = flag_vals [ <NUM_LIT:0> ] <EOL> else : <EOL> flag_val = None <EOL> values_to_rank = [ to_value_type ( x ) for x in <EOL> [ flag_val , env_val_str , config_val_str , kwargs . get ( '<STR_LIT:default>' ) , None ] ] <EOL> ranked_vals = list ( reversed ( list ( RankedValue . prioritized_iter ( * values_to_rank ) ) ) ) <EOL> for ranked_val in ranked_vals : <EOL> if ranked_val . rank == RankedValue . CONFIG : <EOL> details = config_details <EOL> elif ranked_val . rank == RankedValue . ENVIRONMENT : <EOL> details = env_details <EOL> else : <EOL> details = None <EOL> self . _option_tracker . record_option ( scope = self . _scope , <EOL> option = dest , <EOL> value = ranked_val . value , <EOL> rank = ranked_val . rank , <EOL> deprecation_version = kwargs . get ( '<STR_LIT>' ) , <EOL> details = details ) <EOL> def check ( val ) : <EOL> if val is not None : <EOL> choices = kwargs . get ( '<STR_LIT>' ) <EOL> if choices is not None and val not in choices : <EOL> raise ParseError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( val , dest , self . _scope_str ( ) , choices ) ) <EOL> elif kwargs . get ( '<STR_LIT:type>' ) == file_option and not os . path . isfile ( val ) : <EOL> raise ParseError ( '<STR_LIT>' . format ( <EOL> val , dest , self . _scope_str ( ) ) ) <EOL> if is_list_option ( kwargs ) : <EOL> merged_rank = ranked_vals [ - <NUM_LIT:1> ] . rank <EOL> merged_val = ListValueComponent . merge ( <EOL> [ rv . value for rv in ranked_vals if rv . value is not None ] ) . val <EOL> merged_val = [ self . _wrap_type ( kwargs . get ( '<STR_LIT>' , str ) ) ( x ) for x in merged_val ] <EOL> map ( check , merged_val ) <EOL> ret = RankedValue ( merged_rank , merged_val ) <EOL> else : <EOL> ret = ranked_vals [ - <NUM_LIT:1> ] <EOL> check ( ret . value ) <EOL> return ret <EOL> def _inverse_arg ( self , arg ) : <EOL> if arg . startswith ( '<STR_LIT>' ) : <EOL> if arg . startswith ( '<STR_LIT>' ) : <EOL> raise BooleanOptionNameWithNo ( self . scope , arg ) <EOL> return '<STR_LIT>' . format ( arg [ <NUM_LIT:2> : ] ) <EOL> else : <EOL> return None <EOL> def _register_child_parser ( self , child ) : <EOL> self . _child_parsers . append ( child ) <EOL> def _freeze ( self ) : <EOL> self . _frozen = True <EOL> def _scope_str ( self , scope = None ) : <EOL> scope = scope or self . scope <EOL> return '<STR_LIT>' if scope == GLOBAL_SCOPE else "<STR_LIT>" . format ( scope ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . _scope ) </s>
