<s> from __future__ import unicode_literals <EOL> from __future__ import print_function <EOL> import os <EOL> import click <EOL> import traceback <EOL> from types import GeneratorType <EOL> from prompt_toolkit import AbortAction <EOL> from prompt_toolkit import Application <EOL> from prompt_toolkit import CommandLineInterface <EOL> from prompt_toolkit . enums import DEFAULT_BUFFER <EOL> from prompt_toolkit . filters import Always , HasFocus , IsDone <EOL> from prompt_toolkit . layout . processors import HighlightMatchingBracketProcessor , ConditionalProcessor <EOL> from prompt_toolkit . buffer import Buffer <EOL> from prompt_toolkit . interface import AcceptAction <EOL> from prompt_toolkit . shortcuts import create_prompt_layout <EOL> from prompt_toolkit . shortcuts import create_eventloop <EOL> from prompt_toolkit . history import FileHistory <EOL> from . client import DockerClient <EOL> from . client import DockerPermissionException <EOL> from . client import DockerTimeoutException <EOL> from . client import DockerSslException <EOL> from . completer import DockerCompleter <EOL> from . lexer import CommandLexer <EOL> from . formatter import format_data <EOL> from . formatter import output_stream <EOL> from . config import write_default_config , read_config <EOL> from . style import style_factory <EOL> from . keys import get_key_manager <EOL> from . toolbar import create_toolbar_handler <EOL> from . options import OptionError <EOL> from . logger import create_logger <EOL> from . __init__ import __version__ <EOL> click . disable_unicode_literals_warning = True <EOL> class WharfeeCli ( object ) : <EOL> """<STR_LIT>""" <EOL> dcli = None <EOL> keyword_completer = None <EOL> handler = None <EOL> saved_less_opts = None <EOL> config = None <EOL> config_template = '<STR_LIT>' <EOL> config_name = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . config = self . read_configuration ( ) <EOL> self . theme = self . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> log_file = self . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> log_level = self . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . logger = create_logger ( __name__ , log_file , log_level ) <EOL> self . handler = DockerClient ( <EOL> self . config [ '<STR_LIT>' ] . as_int ( '<STR_LIT>' ) , <EOL> self . clear , <EOL> self . set_completer_options ) <EOL> self . completer = DockerCompleter ( <EOL> long_option_names = self . get_long_options ( ) , <EOL> fuzzy = self . get_fuzzy_match ( ) ) <EOL> self . set_completer_options ( ) <EOL> self . saved_less_opts = self . set_less_opts ( ) <EOL> def read_configuration ( self ) : <EOL> """<STR_LIT>""" <EOL> default_config = os . path . join ( <EOL> self . get_package_path ( ) , self . config_template ) <EOL> write_default_config ( default_config , self . config_name ) <EOL> return read_config ( self . config_name , default_config ) <EOL> def get_package_path ( self ) : <EOL> """<STR_LIT>""" <EOL> from wharfee import __file__ as package_root <EOL> return os . path . dirname ( package_root ) <EOL> def set_less_opts ( self ) : <EOL> """<STR_LIT>""" <EOL> opts = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return opts <EOL> def revert_less_opts ( self ) : <EOL> """<STR_LIT>""" <EOL> os . environ [ '<STR_LIT>' ] = self . saved_less_opts <EOL> def write_config_file ( self ) : <EOL> """<STR_LIT>""" <EOL> self . config . write ( ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> click . clear ( ) <EOL> def set_completer_options ( self , cons = True , runs = True , imgs = True , vols = True ) : <EOL> """<STR_LIT>""" <EOL> if cons : <EOL> cs = self . handler . containers ( all = True ) <EOL> if cs and len ( cs ) > <NUM_LIT:0> and isinstance ( cs [ <NUM_LIT:0> ] , dict ) : <EOL> containers = [ name for c in cs for name in c [ '<STR_LIT>' ] ] <EOL> self . completer . set_containers ( containers ) <EOL> if runs : <EOL> cs = self . handler . containers ( ) <EOL> if cs and len ( cs ) > <NUM_LIT:0> and isinstance ( cs [ <NUM_LIT:0> ] , dict ) : <EOL> running = [ name for c in cs for name in c [ '<STR_LIT>' ] ] <EOL> self . completer . set_running ( running ) <EOL> if imgs : <EOL> def format_tagged ( tagname , img_id ) : <EOL> if tagname == '<STR_LIT>' : <EOL> return img_id [ : <NUM_LIT:11> ] <EOL> return tagname <EOL> def parse_image_name ( tag , img_id ) : <EOL> if '<STR_LIT::>' in tag : <EOL> result = tag . split ( '<STR_LIT::>' , <NUM_LIT:2> ) [ <NUM_LIT:0> ] <EOL> else : <EOL> result = tag <EOL> if result == '<STR_LIT>' : <EOL> result = img_id [ : <NUM_LIT:11> ] <EOL> return result <EOL> ims = self . handler . images ( ) <EOL> if ims and len ( ims ) > <NUM_LIT:0> and isinstance ( ims [ <NUM_LIT:0> ] , dict ) : <EOL> images = set ( [ ] ) <EOL> tagged = set ( [ ] ) <EOL> for im in ims : <EOL> repo_tag = '<STR_LIT>' . format ( im [ '<STR_LIT>' ] , im [ '<STR_LIT>' ] ) <EOL> images . add ( parse_image_name ( repo_tag , im [ '<STR_LIT>' ] ) ) <EOL> tagged . add ( format_tagged ( repo_tag , im [ '<STR_LIT>' ] ) ) <EOL> self . completer . set_images ( images ) <EOL> self . completer . set_tagged ( tagged ) <EOL> if vols : <EOL> vs = self . handler . volume_ls ( quiet = True ) <EOL> self . completer . set_volumes ( vs ) <EOL> def set_fuzzy_match ( self , is_fuzzy ) : <EOL> """<STR_LIT>""" <EOL> self . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] = is_fuzzy <EOL> self . completer . set_fuzzy_match ( is_fuzzy ) <EOL> def get_fuzzy_match ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config [ '<STR_LIT>' ] . as_bool ( '<STR_LIT>' ) <EOL> def set_long_options ( self , is_long ) : <EOL> """<STR_LIT>""" <EOL> self . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] = is_long <EOL> self . completer . set_long_options ( is_long ) <EOL> def get_long_options ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config [ '<STR_LIT>' ] . as_bool ( '<STR_LIT>' ) <EOL> def refresh_completions ( self ) : <EOL> """<STR_LIT>""" <EOL> self . set_completer_options ( self . handler . is_refresh_containers , <EOL> self . handler . is_refresh_running , <EOL> self . handler . is_refresh_images , <EOL> self . handler . is_refresh_volumes ) <EOL> def run_cli ( self ) : <EOL> """<STR_LIT>""" <EOL> print ( '<STR_LIT>' , __version__ ) <EOL> print ( '<STR_LIT>' ) <EOL> history = FileHistory ( os . path . expanduser ( '<STR_LIT>' ) ) <EOL> toolbar_handler = create_toolbar_handler ( self . get_long_options , self . get_fuzzy_match ) <EOL> layout = create_prompt_layout ( <EOL> message = '<STR_LIT>' , <EOL> lexer = CommandLexer , <EOL> get_bottom_toolbar_tokens = toolbar_handler , <EOL> extra_input_processors = [ <EOL> ConditionalProcessor ( <EOL> processor = HighlightMatchingBracketProcessor ( <EOL> chars = '<STR_LIT>' ) , <EOL> filter = HasFocus ( DEFAULT_BUFFER ) & ~ IsDone ( ) ) <EOL> ] <EOL> ) <EOL> cli_buffer = Buffer ( <EOL> history = history , <EOL> completer = self . completer , <EOL> complete_while_typing = Always ( ) , <EOL> accept_action = AcceptAction . RETURN_DOCUMENT ) <EOL> manager = get_key_manager ( <EOL> self . set_long_options , <EOL> self . get_long_options , <EOL> self . set_fuzzy_match , <EOL> self . get_fuzzy_match ) <EOL> application = Application ( <EOL> style = style_factory ( self . theme ) , <EOL> layout = layout , <EOL> buffer = cli_buffer , <EOL> key_bindings_registry = manager . registry , <EOL> on_exit = AbortAction . RAISE_EXCEPTION , <EOL> on_abort = AbortAction . RETRY , <EOL> ignore_case = True ) <EOL> eventloop = create_eventloop ( ) <EOL> self . dcli = CommandLineInterface ( <EOL> application = application , <EOL> eventloop = eventloop ) <EOL> while True : <EOL> try : <EOL> document = self . dcli . run ( ) <EOL> self . handler . handle_input ( document . text ) <EOL> if isinstance ( self . handler . output , GeneratorType ) : <EOL> output_stream ( self . handler . command , <EOL> self . handler . output , <EOL> self . handler . logs ) <EOL> elif self . handler . output is not None : <EOL> lines = format_data ( <EOL> self . handler . command , <EOL> self . handler . output ) <EOL> click . echo_via_pager ( '<STR_LIT:\n>' . join ( lines ) ) <EOL> if self . handler . after : <EOL> for line in self . handler . after ( ) : <EOL> click . echo ( line ) <EOL> if self . handler . exception : <EOL> self . logger . warning ( '<STR_LIT>' , <EOL> self . handler . exception ) <EOL> self . refresh_completions ( ) <EOL> except OptionError as ex : <EOL> self . logger . debug ( '<STR_LIT>' , ex ) <EOL> self . logger . error ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> click . secho ( ex . msg , fg = '<STR_LIT>' ) <EOL> except KeyboardInterrupt : <EOL> if self . handler . after : <EOL> click . echo ( '<STR_LIT>' ) <EOL> for line in self . handler . after ( ) : <EOL> click . echo ( line ) <EOL> self . refresh_completions ( ) <EOL> except DockerPermissionException as ex : <EOL> self . logger . debug ( '<STR_LIT>' , ex ) <EOL> self . logger . error ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> click . secho ( ex . message , fg = '<STR_LIT>' ) <EOL> except EOFError : <EOL> break <EOL> except Exception as ex : <EOL> self . logger . debug ( '<STR_LIT>' , ex ) <EOL> self . logger . error ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> click . secho ( "<STR_LIT>" . format ( ex ) , fg = '<STR_LIT>' ) <EOL> break <EOL> self . revert_less_opts ( ) <EOL> self . write_config_file ( ) <EOL> print ( '<STR_LIT>' ) <EOL> @ click . command ( ) <EOL> def cli ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dcli = WharfeeCli ( ) <EOL> dcli . run_cli ( ) <EOL> except DockerTimeoutException as ex : <EOL> click . secho ( ex . message , fg = '<STR_LIT>' ) <EOL> except DockerSslException as ex : <EOL> click . secho ( ex . message , fg = '<STR_LIT>' ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> cli ( ) </s>
