<s> import copy <EOL> import six <EOL> from st2common import log as logging <EOL> from st2common . util import action_db as action_db_util <EOL> from st2common . util . casts import get_cast <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def _merge_param_meta_values ( action_meta = None , runner_meta = None ) : <EOL> runner_meta_keys = runner_meta . keys ( ) if runner_meta else [ ] <EOL> action_meta_keys = action_meta . keys ( ) if action_meta else [ ] <EOL> all_keys = set ( runner_meta_keys ) . union ( set ( action_meta_keys ) ) <EOL> merged_meta = { } <EOL> if runner_meta and runner_meta . get ( '<STR_LIT>' , False ) : <EOL> merged_meta = runner_meta <EOL> for key in all_keys : <EOL> if key in action_meta_keys and key not in runner_meta_keys : <EOL> merged_meta [ key ] = action_meta [ key ] <EOL> elif key in runner_meta_keys and key not in action_meta_keys : <EOL> merged_meta [ key ] = runner_meta [ key ] <EOL> else : <EOL> if key in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> merged_meta [ key ] = runner_meta . get ( key , False ) or action_meta . get ( key , False ) <EOL> else : <EOL> merged_meta [ key ] = action_meta . get ( key ) <EOL> return merged_meta <EOL> def get_params_view ( action_db = None , runner_db = None , merged_only = False ) : <EOL> runner_params = copy . deepcopy ( runner_db . runner_parameters ) if runner_db else { } <EOL> action_params = copy . deepcopy ( action_db . parameters ) if action_db else { } <EOL> parameters = set ( runner_params . keys ( ) ) . union ( set ( action_params . keys ( ) ) ) <EOL> merged_params = { } <EOL> for param in parameters : <EOL> merged_params [ param ] = _merge_param_meta_values ( action_meta = action_params . get ( param ) , <EOL> runner_meta = runner_params . get ( param ) ) <EOL> if merged_only : <EOL> return merged_params <EOL> def is_required ( param_meta ) : <EOL> return param_meta . get ( '<STR_LIT>' , False ) <EOL> def is_immutable ( param_meta ) : <EOL> return param_meta . get ( '<STR_LIT>' , False ) <EOL> immutable = { param for param in parameters if is_immutable ( merged_params . get ( param ) ) } <EOL> required = { param for param in parameters if is_required ( merged_params . get ( param ) ) } <EOL> required = required - immutable <EOL> optional = parameters - required - immutable <EOL> required_params = { k : merged_params [ k ] for k in required } <EOL> optional_params = { k : merged_params [ k ] for k in optional } <EOL> immutable_params = { k : merged_params [ k ] for k in immutable } <EOL> return ( required_params , optional_params , immutable_params ) <EOL> def cast_params ( action_ref , params , cast_overrides = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> params = params or { } <EOL> action_db = action_db_util . get_action_by_ref ( action_ref ) <EOL> if not action_db : <EOL> raise ValueError ( '<STR_LIT>' % ( action_ref ) ) <EOL> action_parameters_schema = action_db . parameters <EOL> runnertype_db = action_db_util . get_runnertype_by_name ( action_db . runner_type [ '<STR_LIT:name>' ] ) <EOL> runner_parameters_schema = runnertype_db . runner_parameters <EOL> parameters_schema = { } <EOL> if runner_parameters_schema : <EOL> parameters_schema . update ( runner_parameters_schema ) <EOL> if action_parameters_schema : <EOL> parameters_schema . update ( action_parameters_schema ) <EOL> for k , v in six . iteritems ( params ) : <EOL> parameter_schema = parameters_schema . get ( k , None ) <EOL> if not parameter_schema : <EOL> LOG . debug ( '<STR_LIT>' , k , v ) <EOL> continue <EOL> parameter_type = parameter_schema . get ( '<STR_LIT:type>' , None ) <EOL> if not parameter_type : <EOL> LOG . debug ( '<STR_LIT>' , k , v ) <EOL> continue <EOL> cast = cast_overrides . get ( parameter_type , None ) if cast_overrides else None <EOL> if not cast : <EOL> cast = get_cast ( cast_type = parameter_type ) <EOL> if not cast : <EOL> LOG . debug ( '<STR_LIT>' , k , v , <EOL> parameter_type ) <EOL> continue <EOL> LOG . debug ( '<STR_LIT>' , v , type ( v ) , parameter_type ) <EOL> try : <EOL> params [ k ] = cast ( v ) <EOL> except Exception : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( v , k , parameter_type ) ) <EOL> raise ValueError ( msg ) <EOL> return params <EOL> def validate_action_parameters ( action_ref , inputs ) : <EOL> input_set = set ( inputs . keys ( ) ) <EOL> parameters = action_db_util . get_action_parameters_specs ( action_ref ) <EOL> required = set ( [ param for param , meta in six . iteritems ( parameters ) <EOL> if meta . get ( '<STR_LIT>' , False ) and '<STR_LIT:default>' not in meta ] ) <EOL> requires = sorted ( required . difference ( input_set ) ) <EOL> unexpected = sorted ( input_set . difference ( set ( parameters . keys ( ) ) ) ) <EOL> return requires , unexpected </s>
