ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: usr / local / jetapps / var / lib / aws / lib / python2.7 / site-packages / awscli / customizations / history

# Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
#     http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import datetime
import json
import sys
import xml.parsers.expat
import xml.dom.minidom

import colorama

from awscli.compat import six
from awscli.customizations.history.commands import HistorySubcommand
from awscli.customizations.history.filters import RegexFilter

class Formatter(object):
    def __init__(self, output=None, include=None, exclude=None):
        """Formats and outputs CLI history events

:type output: File-like obj
        :param output: The stream to write the formatted event to. By default
            sys.stdout is used.

:type include: list
        :param include: A filter specifying which event to only be displayed.
            This parameter is mutually exclusive with exclude.

:type exclude: list
        :param exclude: A filter specifying which events to exclude from being
            displayed. This parameter is mutually exclusive with include.

"""
        self._output = output
        if self._output is None:
            self._output = sys.stdout
        if include and exclude:
            raise ValueError(
                'Either input or exclude can be provided but not both')
        self._include = include
        self._exclude = exclude

def display(self, event_record):
        """Displays a formatted version of the event record

:type event_record: dict
        :param event_record: The event record to format and display.
        """
        if self._should_display(event_record):
            self._display(event_record)

def _display(self, event_record):
        raise NotImplementedError('_display()')

def _should_display(self, event_record):
        if self._include:
            return event_record['event_type'] in self._include
        elif self._exclude:
            return event_record['event_type'] not in self._exclude
        else:
            return True

class DetailedFormatter(Formatter):
    _SIG_FILTER = RegexFilter(
        'Signature=([a-z0-9]{4})[a-z0-9]{60}',
        r'Signature=\1...',
    )

_SECTIONS = {
        'CLI_VERSION': {
            'title': 'AWS CLI command entered',
            'values': [
                {'description': 'with AWS CLI version'}
            ]
        },
        'CLI_ARGUMENTS': {
            'values': [
                {'description': 'with arguments'}
            ]
        },
        'API_CALL': {
            'title': 'API call made',
            'values': [
                {
                    'description': 'to service',
                    'payload_key': 'service'
                },
                {
                    'description': 'using operation',
                    'payload_key': 'operation'
                },
                {
                    'description': 'with parameters',
                    'payload_key': 'params',
                    'value_format': 'dictionary'
                }
            ]
        },
        'HTTP_REQUEST': {
            'title': 'HTTP request sent',
            'values': [
                {
                    'description': 'to URL',
                    'payload_key': 'url'
                },
                {
                    'description': 'with method',
                    'payload_key': 'method'
                },
                {
                    'description': 'with headers',
                    'payload_key': 'headers',
                    'value_format': 'dictionary',
                    'filters': [_SIG_FILTER]
                },
                {
                    'description': 'with body',
                    'payload_key': 'body',
                    'value_format': 'http_body'
                }

]
        },
        'HTTP_RESPONSE': {
            'title': 'HTTP response received',
            'values': [
                {
                    'description': 'with status code',
                    'payload_key': 'status_code'
                },
                {
                    'description': 'with headers',
                    'payload_key': 'headers',
                    'value_format': 'dictionary'
                },
                {
                    'description': 'with body',
                    'payload_key': 'body',
                    'value_format': 'http_body'
                }
            ]
        },
        'PARSED_RESPONSE': {
            'title': 'HTTP response parsed',
            'values': [
                {
                    'description': 'parsed to',
                    'value_format': 'dictionary'
                }
            ]
        },
        'CLI_RC': {
            'title': 'AWS CLI command exited',
            'values': [
                {'description': 'with return code'}
            ]
        },
    }

_COMPONENT_COLORS = {
        'title': colorama.Style.BRIGHT,
        'description': colorama.Fore.CYAN
    }

def __init__(self, output=None, include=None, exclude=None, colorize=True):
        super(DetailedFormatter, self).__init__(output, include, exclude)
        self._request_id_to_api_num = {}
        self._num_api_calls = 0
        self._colorize = colorize
        self._value_pformatter = SectionValuePrettyFormatter()
        if self._colorize:
            colorama.init(autoreset=True, strip=False)

def _display(self, event_record):
        section_definition = self._SECTIONS.get(event_record['event_type'])
        if section_definition is not None:
            self._display_section(event_record, section_definition)

def _display_section(self, event_record, section_definition):
        if 'title' in section_definition:
            self._display_title(section_definition['title'], event_record)
        for value_definition in section_definition['values']:
            self._display_value(value_definition, event_record)

def _display_title(self, title, event_record):
        formatted_title = self._format_section_title(title, event_record)
        self._write_output(formatted_title)

def _display_value(self, value_definition, event_record):
        value_description = value_definition['description']
        event_record_payload = event_record['payload']
        value = event_record_payload
        if 'payload_key' in value_definition:
            value = event_record_payload[value_definition['payload_key']]
        formatted_value = self._format_description(value_description)
        formatted_value += self._format_value(
            value, event_record, value_definition.get('value_format')
        )
        if 'filters' in value_definition:
            for text_filter in value_definition['filters']:
                formatted_value = text_filter.filter_text(formatted_value)
        self._write_output(formatted_value)

def _write_output(self, content):
        if isinstance(content, six.text_type):
            content = content.encode('utf-8')
        self._output.write(content)

def _format_section_title(self, title, event_record):
        formatted_title = title
        api_num = self._get_api_num(event_record)
        if api_num is not None:
            formatted_title = ('[%s] ' % api_num) + formatted_title
        formatted_title = self._color_if_configured(formatted_title, 'title')
        formatted_title += '\n'

formatted_timestamp = self._format_description('at time')
        formatted_timestamp += self._format_value(
            event_record['timestamp'], event_record, value_format='timestamp')

return '\n' + formatted_title + formatted_timestamp

def _get_api_num(self, event_record):
        request_id = event_record['request_id']
        if request_id:
            if request_id not in self._request_id_to_api_num:
                self._request_id_to_api_num[
                    request_id] = self._num_api_calls
                self._num_api_calls += 1
            return self._request_id_to_api_num[request_id]

def _format_description(self, value_description):
        return self._color_if_configured(
            value_description + ': ', 'description')

def _format_value(self, value, event_record, value_format=None):
        if value_format:
            formatted_value = self._value_pformatter.pformat(
                value, value_format, event_record)
        else:
            formatted_value = str(value)
        return formatted_value + '\n'

def _color_if_configured(self, text, component):
        if self._colorize:
            color = self._COMPONENT_COLORS[component]
            return color + text + colorama.Style.RESET_ALL
        return text

class SectionValuePrettyFormatter(object):
    def pformat(self, value, value_format, event_record):
        return getattr(self, '_pformat_' + value_format)(value, event_record)

def _pformat_timestamp(self, event_timestamp, event_record=None):
        return datetime.datetime.fromtimestamp(
            event_timestamp/1000.0).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

def _pformat_dictionary(self, obj, event_record=None):
        return json.dumps(obj=obj, sort_keys=True, indent=4)

def _pformat_http_body(self, body, event_record):
        if not body:
            return 'There is no associated body'
        elif event_record['payload'].get('streaming', False):
            return 'The body is a stream and will not be displayed'
        elif self._is_xml(body):
            # TODO: Figure out a way to minimize the number of times we have
            # to parse the XML. Currently at worst, it will take three times.
            # One to determine if it is XML, another to stip whitespace, and
            # a third to convert to make it pretty. This is an issue as it
            # can cause issues when there are large XML payloads such as
            # an s3 ListObjects call.
            return self._get_pretty_xml(body)
        elif self._is_json_structure(body):
            return self._get_pretty_json(body)
        else:
            return body

def _get_pretty_xml(self, body):
        # The body is parsed and whitespace is stripped because some services
        # like ec2 already return pretty XML and if toprettyxml() was applied
        # to it, it will add even more newlines and spaces on top of it.
        # So this just removes all whitespace from the start to prevent the
        # chance of adding to much newlines and spaces when toprettyxml()
        # is called.
        stripped_body = self._strip_whitespace(body)
        xml_dom = xml.dom.minidom.parseString(stripped_body)
        return xml_dom.toprettyxml(indent=' '*4, newl='\n')

def _get_pretty_json(self, body):
        # The json body is loaded so it can be dumped in a format that
        # is desired.
        obj = json.loads(body)
        return self._pformat_dictionary(obj)

def _is_xml(self, body):
        try:
            xml.dom.minidom.parseString(body)
        except xml.parsers.expat.ExpatError:
            return False
        return True

def _strip_whitespace(self, xml_string):
        xml_dom = xml.dom.minidom.parseString(xml_string)
        return ''.join(
            [line.strip() for line in xml_dom.toxml().splitlines()]
        )

def _is_json_structure(self, body):
        if body.startswith('{'):
            try:
                json.loads(body)
                return True
            except json.decoder.JSONDecodeError:
                return False
        return False

class ShowCommand(HistorySubcommand):
    NAME = 'show'
    DESCRIPTION = (
        'Shows the various events related to running a specific CLI command. '
        'If this command is ran without any positional arguments, it will '
        'display the events for the last CLI command ran.'
    )
    FORMATTERS = {
        'detailed': DetailedFormatter
    }
    ARG_TABLE = [
        {'name': 'command_id', 'nargs': '?', 'default': 'latest',
         'positional_arg': True,
         'help_text': (
             'The ID of the CLI command to show. If this positional argument '
             'is omitted, it will show the last the CLI command ran.')},
        {'name': 'include', 'nargs': '+',
         'help_text': (
             'Specifies which events to **only** include when showing the '
             'CLI command. This argument is mutually exclusive with '
             '``--exclude``.')},
        {'name': 'exclude', 'nargs': '+',
         'help_text': (
             'Specifies which events to exclude when showing the '
             'CLI command. This argument is mutually exclusive with '
             '``--include``.')},
        {'name': 'format', 'choices': FORMATTERS.keys(),
         'default': 'detailed', 'help_text': (
            'Specifies which format to use in showing the events for '
            'the specified CLI command. The following formats are '
            'supported:\n\n'
            '<ul>'
            '<li> detailed - This the default format. It prints out a '
            'detailed overview of the CLI command ran. It displays all '
            'of the key events in the command lifecycle where each '
            'important event has a title and its important values '
            'underneath. The events are ordered by timestamp and events of '
            'the same API call are associated together with the '
            '[``api_id``] notation where events that share the same '
            '``api_id`` belong to the lifecycle of the same API call.'
            '</li>'
            '</ul>'
            )
         }
    ]

def _run_main(self, parsed_args, parsed_globals):
        self._connect_to_history_db()
        try:
            self._validate_args(parsed_args)
            with self._get_output_stream() as output_stream:
                formatter = self._get_formatter(
                    parsed_args, parsed_globals, output_stream)
                for record in self._get_record_iterator(parsed_args):
                    formatter.display(record)
        finally:
            self._close_history_db()
        return 0

def _validate_args(self, parsed_args):
        if parsed_args.exclude and parsed_args.include:
            raise ValueError(
                'Either --exclude or --include can be provided but not both')

def _get_formatter(self, parsed_args, parsed_globals, output_stream):
        format_type = parsed_args.format
        formatter_kwargs = {
            'include': parsed_args.include,
            'exclude': parsed_args.exclude,
            'output': output_stream
        }
        if format_type == 'detailed':
            formatter_kwargs['colorize'] = self._should_use_color(
                parsed_globals)
        return self.FORMATTERS[format_type](**formatter_kwargs)

def _get_record_iterator(self, parsed_args):
        if parsed_args.command_id == 'latest':
            return self._db_reader.iter_latest_records()
        else:
            return self._db_reader.iter_records(parsed_args.command_id)

[-] commands.py [edit]
[-] commands.pyc [edit]
[+] ..
[-] db.py [edit]
[-] filters.pyc [edit]
[-] __init__.pyc [edit]
[-] list.py [edit]
[-] filters.py [edit]
[-] constants.py [edit]
[-] list.pyc [edit]
[-] show.pyc [edit]
[-] __init__.py [edit]
[-] show.py [edit]
[-] constants.pyc [edit]
[-] db.pyc [edit]