# Copyright 2013 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 base64

import contextlib

import csv

import datetime

import os

import signal

import subprocess

import sys

from awscli.compat import (

StringIO,

get_binary_stdout,

get_popen_kwargs_for_pager_cmd,

)

def split_on_commas(value):

if not any(char in value for char in ['"', '\\', "'", ']', '[']):

# No quotes or escaping, just use a simple split.

return value.split(',')

elif not any(char in value for char in ['"', "'", '[', ']']):

# Simple escaping, let the csv module handle it.

return list(csv.reader(StringIO(value), escapechar='\\'))[0]

else:

# If there's quotes for the values, we have to handle this

# ourselves.

return _split_with_quotes(value)

def _split_with_quotes(value):

try:

parts = list(csv.reader(StringIO(value), escapechar='\\'))[0]

except csv.Error:

raise ValueError("Bad csv value: %s" % value)

iter_parts = iter(parts)

new_parts = []

for part in iter_parts:

# Find the first quote

quote_char = _find_quote_char_in_part(part)

# Find an opening list bracket

list_start = part.find('=[')

if (

list_start >= 0

and value.find(']') != -1

and (quote_char is None or part.find(quote_char) > list_start)

):

# This is a list, eat all the items until the end

if ']' in part:

# Short circuit for only one item

new_chunk = part

else:

new_chunk = _eat_items(value, iter_parts, part, ']')

list_items = _split_with_quotes(new_chunk[list_start + 2 : -1])

new_chunk = new_chunk[: list_start + 1] + ','.join(list_items)

new_parts.append(new_chunk)

continue

elif quote_char is None:

new_parts.append(part)

continue

elif part.count(quote_char) == 2:

# Starting and ending quote are in this part.

# While it's not needed right now, this will

# break down if we ever need to escape quotes while

# quoting a value.

new_parts.append(part.replace(quote_char, ''))

continue

# Now that we've found a starting quote char, we

# need to combine the parts until we encounter an end quote.

new_chunk = _eat_items(value, iter_parts, part, quote_char, quote_char)

new_parts.append(new_chunk)

return new_parts

def _eat_items(value, iter_parts, part, end_char, replace_char=''):

"""

Eat items from an iterator, optionally replacing characters with

a blank and stopping when the end_char has been reached.

"""

current = part

chunks = [current.replace(replace_char, '')]

while True:

try:

current = next(iter_parts)

except StopIteration:

raise ValueError(value)

chunks.append(current.replace(replace_char, ''))

if current.endswith(end_char):

break

return ','.join(chunks)

def _find_quote_char_in_part(part):

"""

Returns a single or double quote character, whichever appears first in the

given string. None is returned if the given string doesn't have a single or

double quote character.

"""

quote_char = None

for ch in part:

if ch in ('"', "'"):

quote_char = ch

break

return quote_char

def find_service_and_method_in_event_name(event_name):

"""

Grabs the service id and the operation name from an event name.

This is making the assumption that the event name is in the form

event.service.operation.

"""

split_event = event_name.split('.')[1:]

service_name = None

if len(split_event) > 0:

service_name = split_event[0]

operation_name = None

if len(split_event) > 1:

operation_name = split_event[1]

return service_name, operation_name

def resolve_v2_debug_mode(args):

# Resolve whether v2-debug mode is enabled,

# following the correct precedence order.

if args is None:

return False

if getattr(args, 'v2_debug', False):

return True

if os.environ.get('AWS_CLI_UPGRADE_DEBUG_MODE', '').lower() == 'true':

return True

return False

def is_document_type(shape):

"""Check if shape is a document type"""

return getattr(shape, 'is_document_type', False)

def is_document_type_container(shape):

"""Check if the shape is a document type or wraps document types

This is helpful to determine if a shape purely deals with document types

whether the shape is a document type or it is lists or maps whose base

values are document types.

"""

if not shape:

return False

recording_visitor = ShapeRecordingVisitor()

ShapeWalker().walk(shape, recording_visitor)

end_shape = recording_visitor.visited.pop()

if not is_document_type(end_shape):

return False

for shape in recording_visitor.visited:

if shape.type_name not in ['list', 'map']:

return False

return True

def is_streaming_blob_type(shape):

"""Check if the shape is a streaming blob type."""

return (

shape

and shape.type_name == 'blob'

and shape.serialization.get('streaming', False)

)

def is_tagged_union_type(shape):

"""Check if the shape is a tagged union structure."""

return getattr(shape, 'is_tagged_union', False)

def operation_uses_document_types(operation_model):

"""Check if document types are ever used in the operation"""

recording_visitor = ShapeRecordingVisitor()

walker = ShapeWalker()

walker.walk(operation_model.input_shape, recording_visitor)

walker.walk(operation_model.output_shape, recording_visitor)

for visited_shape in recording_visitor.visited:

if is_document_type(visited_shape):

return True

return False

def json_encoder(obj):

"""JSON encoder that formats datetimes as ISO8601 format

and encodes bytes to UTF-8 Base64 string."""

if isinstance(obj, datetime.datetime):

return obj.isoformat()

elif isinstance(obj, bytes):

return base64.b64encode(obj).decode("utf-8")

else:

raise TypeError('Encountered unrecognized type in JSON encoder.')

@contextlib.contextmanager

def ignore_ctrl_c():

original = signal.signal(signal.SIGINT, signal.SIG_IGN)

try:

yield

finally:

signal.signal(signal.SIGINT, original)

def emit_top_level_args_parsed_event(session, args, remaining=None):

session.emit(

'top-level-args-parsed',

parsed_args=args,

session=session,

remaining_args=remaining,

)

def is_a_tty():

try:

return os.isatty(sys.stdout.fileno())

except Exception:

return False

class OutputStreamFactory:

def __init__(self, popen=None):

self._popen = popen

if popen is None:

self._popen = subprocess.Popen

@contextlib.contextmanager

def get_pager_stream(self, preferred_pager=None):

popen_kwargs = self._get_process_pager_kwargs(preferred_pager)

try:

process = self._popen(**popen_kwargs)

yield process.stdin

except OSError:

# Ignore IOError since this can commonly be raised when a pager

# is closed abruptly and causes a broken pipe.

pass

finally:

process.communicate()

@contextlib.contextmanager

def get_stdout_stream(self):

yield get_binary_stdout()

def _get_process_pager_kwargs(self, pager_cmd):

kwargs = get_popen_kwargs_for_pager_cmd(pager_cmd)

kwargs['stdin'] = subprocess.PIPE

return kwargs

def write_exception(ex, outfile):

outfile.write("\n")

outfile.write(str(ex))

outfile.write("\n")

class ShapeWalker:

def walk(self, shape, visitor):

"""Walk through and visit shapes for introspection

:type shape: botocore.model.Shape

:param shape: Shape to walk

:type visitor: BaseShapeVisitor

:param visitor: The visitor to call when walking a shape

"""

if shape is None:

return

stack = []

return self._walk(shape, visitor, stack)

def _walk(self, shape, visitor, stack):

if shape.name in stack:

return

stack.append(shape.name)

getattr(self, '_walk_%s' % shape.type_name, self._default_scalar_walk)(

shape, visitor, stack

)

stack.pop()

def _walk_structure(self, shape, visitor, stack):

self._do_shape_visit(shape, visitor)

for _, member_shape in shape.members.items():

self._walk(member_shape, visitor, stack)

def _walk_list(self, shape, visitor, stack):

self._do_shape_visit(shape, visitor)

self._walk(shape.member, visitor, stack)

def _walk_map(self, shape, visitor, stack):

self._do_shape_visit(shape, visitor)

self._walk(shape.value, visitor, stack)

def _default_scalar_walk(self, shape, visitor, stack):

self._do_shape_visit(shape, visitor)

def _do_shape_visit(self, shape, visitor):

visitor.visit_shape(shape)

class BaseShapeVisitor:

"""Visit shape encountered by ShapeWalker"""

def visit_shape(self, shape):

pass

class ShapeRecordingVisitor(BaseShapeVisitor):

"""Record shapes visited by ShapeWalker"""

def __init__(self):

self.visited = []

def visit_shape(self, shape):

self.visited.append(shape)

# NOTE: The following interfaces are considered private and are subject

# to abrupt breaking changes. Please do not use them directly.

try:

from botocore.utils import create_nested_client as create_client

except ImportError:

def create_client(session, service_name, **kwargs):

return session.create_client(service_name, **kwargs)

def create_nested_client(session, service_name, **kwargs):

return create_client(session, service_name, **kwargs)