<s> """<STR_LIT>""" <EOL> import os <EOL> import logging <EOL> from collections import namedtuple <EOL> import jmespath <EOL> from botocore import xform_name <EOL> from botocore . exceptions import BotoCoreError <EOL> LOG = logging . getLogger ( __name__ ) <EOL> ServerCompletion = namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:path>' ] ) <EOL> def extract_field_from_jmespath ( expression ) : <EOL> result = jmespath . compile ( expression ) <EOL> current = result . parsed <EOL> while current [ '<STR_LIT>' ] : <EOL> current = current [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> if current [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> return current [ '<STR_LIT:value>' ] <EOL> class ResourceIndexBuilder ( object ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def build_index ( self , resource_data ) : <EOL> index = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : { } , <EOL> } <EOL> service = resource_data [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in service : <EOL> for model in service [ '<STR_LIT>' ] . values ( ) : <EOL> resource_name = model [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] <EOL> for identifier in model [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> first_identifier = model [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> index [ '<STR_LIT>' ] [ resource_name ] = { <EOL> '<STR_LIT>' : model [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> first_identifier [ '<STR_LIT:target>' ] : first_identifier [ '<STR_LIT:path>' ] <EOL> } <EOL> } <EOL> for resource_name , model in resource_data [ '<STR_LIT>' ] . items ( ) : <EOL> if resource_name not in index [ '<STR_LIT>' ] : <EOL> continue <EOL> if '<STR_LIT>' in model : <EOL> resource_actions = model [ '<STR_LIT>' ] <EOL> for action_model in resource_actions . values ( ) : <EOL> op_name = action_model [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> current = { } <EOL> index [ '<STR_LIT>' ] [ op_name ] = current <EOL> for param in action_model [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> if param [ '<STR_LIT:source>' ] == '<STR_LIT>' : <EOL> field_name = extract_field_from_jmespath ( <EOL> param [ '<STR_LIT:target>' ] ) <EOL> current [ field_name ] = { <EOL> '<STR_LIT>' : resource_name , <EOL> '<STR_LIT>' : param [ '<STR_LIT:name>' ] , <EOL> } <EOL> return index <EOL> class CompleterDescriber ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , resource_index ) : <EOL> self . _index = resource_index <EOL> def describe_autocomplete ( self , service , operation , param ) : <EOL> """<STR_LIT>""" <EOL> service_index = self . _index [ service ] <EOL> LOG . debug ( service_index ) <EOL> if param not in service_index . get ( '<STR_LIT>' , { } ) . get ( operation , { } ) : <EOL> LOG . debug ( "<STR_LIT>" , param ) <EOL> return None <EOL> p = service_index [ '<STR_LIT>' ] [ operation ] [ param ] <EOL> resource_name = p [ '<STR_LIT>' ] <EOL> resource_identifier = p [ '<STR_LIT>' ] <EOL> resource_index = service_index [ '<STR_LIT>' ] [ resource_name ] <EOL> completion_operation = resource_index [ '<STR_LIT>' ] <EOL> path = resource_index [ '<STR_LIT>' ] [ resource_identifier ] <EOL> return ServerCompletion ( service = service , operation = completion_operation , <EOL> params = { } , path = path ) <EOL> class CachedClientCreator ( object ) : <EOL> def __init__ ( self , session ) : <EOL> self . _session = session <EOL> self . _client_cache = { } <EOL> def create_client ( self , service_name ) : <EOL> if service_name not in self . _client_cache : <EOL> client = self . _session . create_client ( service_name ) <EOL> self . _client_cache [ service_name ] = client <EOL> return self . _client_cache [ service_name ] <EOL> class CompleterDescriberCreator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , loader ) : <EOL> self . _loader = loader <EOL> self . _describer_cache = { } <EOL> self . _services_with_completions = None <EOL> def create_completer_query ( self , service_name ) : <EOL> """<STR_LIT>""" <EOL> if service_name not in self . _describer_cache : <EOL> query = self . _create_completer_query ( service_name ) <EOL> self . _describer_cache [ service_name ] = query <EOL> return self . _describer_cache [ service_name ] <EOL> def _create_completer_query ( self , service_name ) : <EOL> completions_model = self . _loader . load_service_model ( <EOL> service_name , '<STR_LIT>' ) <EOL> cq = CompleterDescriber ( { service_name : completions_model } ) <EOL> return cq <EOL> def services_with_completions ( self ) : <EOL> if self . _services_with_completions is not None : <EOL> return self . _services_with_completions <EOL> self . _services_with_completions = set ( <EOL> self . _loader . list_available_services ( type_name = '<STR_LIT>' ) ) <EOL> return self . _services_with_completions <EOL> class ServerSideCompleter ( object ) : <EOL> def __init__ ( self , client_creator , describer_creator ) : <EOL> self . _client_creator = client_creator <EOL> self . _describer_creator = describer_creator <EOL> def retrieve_candidate_values ( self , service , operation , param ) : <EOL> """<STR_LIT>""" <EOL> if service not in self . _describer_creator . services_with_completions ( ) : <EOL> return [ ] <EOL> try : <EOL> client = self . _client_creator . create_client ( service ) <EOL> except BotoCoreError as e : <EOL> LOG . debug ( "<STR_LIT>" , <EOL> service , exc_info = True ) <EOL> return [ ] <EOL> api_operation_name = client . meta . method_to_api_mapping . get ( <EOL> operation . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) ) <EOL> if api_operation_name is None : <EOL> return [ ] <EOL> completer = self . _describer_creator . create_completer_query ( service ) <EOL> result = completer . describe_autocomplete ( <EOL> service , api_operation_name , param ) <EOL> if result is None : <EOL> return <EOL> try : <EOL> response = getattr ( client , xform_name ( result . operation , '<STR_LIT:_>' ) ) ( ) <EOL> except Exception as e : <EOL> LOG . debug ( "<STR_LIT>" , service , <EOL> result . operation , e , exc_info = True ) <EOL> return <EOL> results = jmespath . search ( result . path , response ) <EOL> return results <EOL> def main ( ) : <EOL> import sys <EOL> import json <EOL> import os <EOL> import boto3 . session <EOL> data_dir = os . path . join ( <EOL> os . path . dirname ( os . path . dirname ( os . path . abspath ( __file__ ) ) ) , <EOL> '<STR_LIT:data>' ) <EOL> if not os . path . isdir ( data_dir ) : <EOL> os . makedirs ( data_dir ) <EOL> session = boto3 . session . Session ( ) <EOL> loader = session . _loader <EOL> builder = ResourceIndexBuilder ( ) <EOL> for resource_name in session . get_available_resources ( ) : <EOL> api_version = loader . determine_latest_version ( <EOL> resource_name , '<STR_LIT>' ) <EOL> model = loader . load_service_model ( resource_name , '<STR_LIT>' , <EOL> api_version ) <EOL> index = builder . build_index ( model ) <EOL> output_file = os . path . join ( data_dir , resource_name , api_version , <EOL> '<STR_LIT>' ) <EOL> if not os . path . isdir ( os . path . dirname ( output_file ) ) : <EOL> os . makedirs ( os . path . dirname ( output_file ) ) <EOL> with open ( output_file , '<STR_LIT:w>' ) as f : <EOL> f . write ( json . dumps ( index , indent = <NUM_LIT:2> ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
