<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> import os <EOL> import subprocess <EOL> import logging <EOL> import sys <EOL> from prompt_toolkit . document import Document <EOL> from prompt_toolkit . shortcuts import create_eventloop <EOL> from prompt_toolkit . buffer import Buffer <EOL> from prompt_toolkit . filters import Always <EOL> from prompt_toolkit . interface import CommandLineInterface , Application <EOL> from prompt_toolkit . interface import AbortAction , AcceptAction <EOL> from prompt_toolkit . utils import Callback <EOL> from prompt_toolkit . auto_suggest import AutoSuggestFromHistory <EOL> from prompt_toolkit . history import InMemoryHistory , FileHistory <EOL> from awsshell . ui import create_default_layout <EOL> from awsshell . config import Config <EOL> from awsshell . keys import KeyManager <EOL> from awsshell . style import StyleFactory <EOL> from awsshell . toolbar import Toolbar <EOL> from awsshell . utils import build_config_file_path , temporary_file <EOL> from awsshell import compat <EOL> LOG = logging . getLogger ( __name__ ) <EOL> EXIT_REQUESTED = object ( ) <EOL> def create_aws_shell ( completer , model_completer , docs ) : <EOL> return AWSShell ( completer , model_completer , docs ) <EOL> class InputInterrupt ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class ChangeDirHandler ( object ) : <EOL> def __init__ ( self , output = sys . stdout , err = sys . stderr , chdir = os . chdir ) : <EOL> self . _output = output <EOL> self . _err = err <EOL> self . _chdir = chdir <EOL> def run ( self , command , application ) : <EOL> if len ( command ) != <NUM_LIT:2> : <EOL> self . _err . write ( "<STR_LIT>" ) <EOL> return <EOL> dirname = os . path . expandvars ( os . path . expanduser ( command [ <NUM_LIT:1> ] ) ) <EOL> try : <EOL> self . _chdir ( dirname ) <EOL> except OSError as e : <EOL> self . _err . write ( "<STR_LIT>" % e ) <EOL> class EditHandler ( object ) : <EOL> def __init__ ( self , popen_cls = None , env = None , err = sys . stderr ) : <EOL> if popen_cls is None : <EOL> popen_cls = subprocess . Popen <EOL> self . _popen_cls = popen_cls <EOL> if env is None : <EOL> env = os . environ <EOL> self . _env = env <EOL> self . _err = err <EOL> def _get_editor_command ( self ) : <EOL> if '<STR_LIT>' in self . _env : <EOL> return self . _env [ '<STR_LIT>' ] <EOL> else : <EOL> return compat . default_editor ( ) <EOL> def run ( self , command , application ) : <EOL> """<STR_LIT>""" <EOL> all_commands = '<STR_LIT:\n>' . join ( <EOL> [ '<STR_LIT>' + h for h in list ( application . history ) <EOL> if not h . startswith ( ( '<STR_LIT:.>' , '<STR_LIT:!>' ) ) ] ) <EOL> with temporary_file ( '<STR_LIT:w>' ) as f : <EOL> f . write ( all_commands ) <EOL> f . flush ( ) <EOL> editor = self . _get_editor_command ( ) <EOL> try : <EOL> p = self . _popen_cls ( [ editor , f . name ] ) <EOL> p . communicate ( ) <EOL> except OSError : <EOL> self . _err . write ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % editor ) <EOL> class ProfileHandler ( object ) : <EOL> USAGE = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> def __init__ ( self , output = sys . stdout , err = sys . stderr ) : <EOL> self . _output = output <EOL> self . _err = err <EOL> def run ( self , command , application ) : <EOL> """<STR_LIT>""" <EOL> if len ( command ) == <NUM_LIT:1> : <EOL> profile = application . profile <EOL> if profile is None : <EOL> self . _output . write ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> self . _output . write ( "<STR_LIT>" % profile ) <EOL> elif len ( command ) == <NUM_LIT:2> : <EOL> new_profile_name = command [ <NUM_LIT:1> ] <EOL> application . profile = new_profile_name <EOL> self . _output . write ( "<STR_LIT>" % <EOL> new_profile_name ) <EOL> else : <EOL> self . _err . write ( "<STR_LIT>" % self . USAGE ) <EOL> class ExitHandler ( object ) : <EOL> def run ( self , command , application ) : <EOL> return EXIT_REQUESTED <EOL> class DotCommandHandler ( object ) : <EOL> HANDLER_CLASSES = { <EOL> '<STR_LIT>' : EditHandler , <EOL> '<STR_LIT>' : ProfileHandler , <EOL> '<STR_LIT>' : ChangeDirHandler , <EOL> '<STR_LIT>' : ExitHandler , <EOL> '<STR_LIT>' : ExitHandler , <EOL> } <EOL> def __init__ ( self , output = sys . stdout , err = sys . stderr ) : <EOL> self . _output = output <EOL> self . _err = err <EOL> def handle_cmd ( self , command , application ) : <EOL> """<STR_LIT>""" <EOL> parts = command . split ( ) <EOL> cmd_name = parts [ <NUM_LIT:0> ] [ <NUM_LIT:1> : ] <EOL> if cmd_name not in self . HANDLER_CLASSES : <EOL> self . _unknown_cmd ( parts , application ) <EOL> else : <EOL> return self . HANDLER_CLASSES [ cmd_name ] ( ) . run ( parts , application ) <EOL> def _unknown_cmd ( self , cmd_parts , application ) : <EOL> self . _err . write ( "<STR_LIT>" % cmd_parts [ <NUM_LIT:0> ] ) <EOL> class AWSShell ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , completer , model_completer , docs ) : <EOL> self . completer = completer <EOL> self . model_completer = model_completer <EOL> self . history = InMemoryHistory ( ) <EOL> self . file_history = FileHistory ( build_config_file_path ( '<STR_LIT>' ) ) <EOL> self . _cli = None <EOL> self . _docs = docs <EOL> self . current_docs = u'<STR_LIT>' <EOL> self . refresh_cli = False <EOL> self . key_manager = None <EOL> self . _dot_cmd = DotCommandHandler ( ) <EOL> self . _env = os . environ . copy ( ) <EOL> self . _profile = None <EOL> self . config_obj = None <EOL> self . config_section = None <EOL> self . enable_vi_bindings = None <EOL> self . show_completion_columns = None <EOL> self . show_help = None <EOL> self . theme = None <EOL> self . load_config ( ) <EOL> def load_config ( self ) : <EOL> """<STR_LIT>""" <EOL> config = Config ( ) <EOL> self . config_obj = config . load ( '<STR_LIT>' ) <EOL> self . config_section = self . config_obj [ '<STR_LIT>' ] <EOL> self . model_completer . match_fuzzy = self . config_section . as_bool ( <EOL> '<STR_LIT>' ) <EOL> self . enable_vi_bindings = self . config_section . as_bool ( <EOL> '<STR_LIT>' ) <EOL> self . show_completion_columns = self . config_section . as_bool ( <EOL> '<STR_LIT>' ) <EOL> self . show_help = self . config_section . as_bool ( '<STR_LIT>' ) <EOL> self . theme = self . config_section [ '<STR_LIT>' ] <EOL> def save_config ( self ) : <EOL> """<STR_LIT>""" <EOL> self . config_section [ '<STR_LIT>' ] = self . model_completer . match_fuzzy <EOL> self . config_section [ '<STR_LIT>' ] = self . enable_vi_bindings <EOL> self . config_section [ '<STR_LIT>' ] = self . show_completion_columns <EOL> self . config_section [ '<STR_LIT>' ] = self . show_help <EOL> self . config_section [ '<STR_LIT>' ] = self . theme <EOL> self . config_obj . write ( ) <EOL> @ property <EOL> def cli ( self ) : <EOL> if self . _cli is None or self . refresh_cli : <EOL> self . _cli = self . create_cli_interface ( self . show_completion_columns ) <EOL> self . refresh_cli = False <EOL> return self . _cli <EOL> def run ( self ) : <EOL> while True : <EOL> try : <EOL> document = self . cli . run ( ) <EOL> text = document . text <EOL> except InputInterrupt : <EOL> pass <EOL> except ( KeyboardInterrupt , EOFError ) : <EOL> self . save_config ( ) <EOL> break <EOL> else : <EOL> if text . startswith ( '<STR_LIT:.>' ) : <EOL> result = self . _dot_cmd . handle_cmd ( text , application = self ) <EOL> if result is EXIT_REQUESTED : <EOL> break <EOL> else : <EOL> if text . startswith ( '<STR_LIT:!>' ) : <EOL> full_cmd = text [ <NUM_LIT:1> : ] <EOL> else : <EOL> full_cmd = '<STR_LIT>' + text <EOL> self . history . append ( full_cmd ) <EOL> self . current_docs = u'<STR_LIT>' <EOL> self . cli . buffers [ '<STR_LIT>' ] . reset ( <EOL> initial_document = Document ( self . current_docs , <EOL> cursor_position = <NUM_LIT:0> ) ) <EOL> self . cli . request_redraw ( ) <EOL> p = subprocess . Popen ( full_cmd , shell = True , env = self . _env ) <EOL> p . communicate ( ) <EOL> def stop_input_and_refresh_cli ( self ) : <EOL> """<STR_LIT>""" <EOL> self . refresh_cli = True <EOL> self . cli . request_redraw ( ) <EOL> raise InputInterrupt <EOL> def create_layout ( self , display_completions_in_columns , toolbar ) : <EOL> from awsshell . lexer import ShellLexer <EOL> lexer = ShellLexer <EOL> if self . config_section [ '<STR_LIT>' ] == '<STR_LIT:none>' : <EOL> lexer = None <EOL> return create_default_layout ( <EOL> self , u'<STR_LIT>' , lexer = lexer , reserve_space_for_menu = True , <EOL> display_completions_in_columns = display_completions_in_columns , <EOL> get_bottom_toolbar_tokens = toolbar . handler ) <EOL> def create_buffer ( self , completer , history ) : <EOL> return Buffer ( <EOL> history = history , <EOL> auto_suggest = AutoSuggestFromHistory ( ) , <EOL> enable_history_search = True , <EOL> completer = completer , <EOL> complete_while_typing = Always ( ) , <EOL> accept_action = AcceptAction . RETURN_DOCUMENT ) <EOL> def create_key_manager ( self ) : <EOL> """<STR_LIT>""" <EOL> def set_match_fuzzy ( match_fuzzy ) : <EOL> """<STR_LIT>""" <EOL> self . model_completer . match_fuzzy = match_fuzzy <EOL> def set_enable_vi_bindings ( enable_vi_bindings ) : <EOL> """<STR_LIT>""" <EOL> self . enable_vi_bindings = enable_vi_bindings <EOL> def set_show_completion_columns ( show_completion_columns ) : <EOL> """<STR_LIT>""" <EOL> self . show_completion_columns = show_completion_columns <EOL> def set_show_help ( show_help ) : <EOL> """<STR_LIT>""" <EOL> self . show_help = show_help <EOL> return KeyManager ( <EOL> lambda : self . model_completer . match_fuzzy , set_match_fuzzy , <EOL> lambda : self . enable_vi_bindings , set_enable_vi_bindings , <EOL> lambda : self . show_completion_columns , set_show_completion_columns , <EOL> lambda : self . show_help , set_show_help , <EOL> self . stop_input_and_refresh_cli ) <EOL> def create_application ( self , completer , history , <EOL> display_completions_in_columns ) : <EOL> self . key_manager = self . create_key_manager ( ) <EOL> toolbar = Toolbar ( <EOL> lambda : self . model_completer . match_fuzzy , <EOL> lambda : self . enable_vi_bindings , <EOL> lambda : self . show_completion_columns , <EOL> lambda : self . show_help ) <EOL> style_factory = StyleFactory ( self . theme ) <EOL> buffers = { <EOL> '<STR_LIT>' : Buffer ( read_only = True ) <EOL> } <EOL> return Application ( <EOL> layout = self . create_layout ( display_completions_in_columns , toolbar ) , <EOL> mouse_support = False , <EOL> style = style_factory . style , <EOL> buffers = buffers , <EOL> buffer = self . create_buffer ( completer , history ) , <EOL> on_abort = AbortAction . RETRY , <EOL> on_exit = AbortAction . RAISE_EXCEPTION , <EOL> on_input_timeout = Callback ( self . on_input_timeout ) , <EOL> key_bindings_registry = self . key_manager . manager . registry , <EOL> ) <EOL> def on_input_timeout ( self , cli ) : <EOL> if not self . show_help : <EOL> return <EOL> document = cli . current_buffer . document <EOL> text = document . text <EOL> LOG . debug ( "<STR_LIT>" , text ) <EOL> LOG . debug ( "<STR_LIT>" , self . completer . current_command ) <EOL> if text . strip ( ) : <EOL> command = self . completer . current_command <EOL> key_name = '<STR_LIT:.>' . join ( command . split ( ) ) . encode ( '<STR_LIT:utf-8>' ) <EOL> last_option = self . completer . last_option <EOL> if last_option : <EOL> self . current_docs = self . _docs . extract_param ( <EOL> key_name , last_option ) <EOL> else : <EOL> self . current_docs = self . _docs . extract_description ( key_name ) <EOL> else : <EOL> self . current_docs = u'<STR_LIT>' <EOL> cli . buffers [ '<STR_LIT>' ] . reset ( <EOL> initial_document = Document ( self . current_docs , cursor_position = <NUM_LIT:0> ) ) <EOL> cli . request_redraw ( ) <EOL> def create_cli_interface ( self , display_completions_in_columns ) : <EOL> loop = create_eventloop ( ) <EOL> app = self . create_application ( self . completer , <EOL> self . file_history , <EOL> display_completions_in_columns ) <EOL> cli = CommandLineInterface ( application = app , eventloop = loop ) <EOL> return cli <EOL> @ property <EOL> def profile ( self ) : <EOL> return self . _profile <EOL> @ profile . setter <EOL> def profile ( self , new_profile_name ) : <EOL> self . _env [ '<STR_LIT>' ] = new_profile_name <EOL> self . completer . change_profile ( new_profile_name ) <EOL> self . _profile = new_profile_name </s>
