aws-cli/awscli/alias.py at develop · rubythonode/aws-cli · GitHub

History

282 lines (235 loc) · 10.9 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

# 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 logging

import os

import shlex

import subprocess

from botocore.configloader import raw_config_parse

from awscli.commands import CLICommand

from awscli.utils import emit_top_level_args_parsed_event

LOG = logging.getLogger(__name__)

class InvalidAliasException(Exception):

pass

class AliasLoader(object):

def __init__(self,

alias_filename=os.path.expanduser(

os.path.join('~', '.aws', 'cli', 'alias'))):

"""Interface for loading and interacting with alias file

:param alias_filename: The name of the file to load aliases from.

This file must be an INI file.

"""

self._filename = alias_filename

self._aliases = None

def _build_aliases(self):

self._aliases = self._load_aliases()

self._cleanup_alias_values(self._aliases.get('toplevel', {}))

def _load_aliases(self):

if os.path.exists(self._filename):

return raw_config_parse(

self._filename, parse_subsections=False)

return {'toplevel': {}}

def _cleanup_alias_values(self, aliases):

for alias in aliases:

# Beginning and end line separators should not be included

# in the internal representation of the alias value.

aliases[alias] = aliases[alias].strip()

def get_aliases(self):

if self._aliases is None:

self._build_aliases()

return self._aliases.get('toplevel', {})

class AliasCommandInjector(object):

def __init__(self, session, alias_loader):

"""Injects alias commands for a command table

:type session: botocore.session.Session

:param session: The botocore session

:type alias_loader: awscli.alias.AliasLoader

:param alias_loader: The alias loader to use

"""

self._session = session

self._alias_loader = alias_loader

def inject_aliases(self, command_table, parser):

for alias_name, alias_value in \

self._alias_loader.get_aliases().items():

if alias_value.startswith('!'):

alias_cmd = ExternalAliasCommand(alias_name, alias_value)

else:

service_alias_cmd_args = [

alias_name, alias_value, self._session, command_table,

parser

]

# If the alias name matches something already in the

# command table provide the command it is about

# to clobber as a possible reference that it will

# need to proxy to.

if alias_name in command_table:

service_alias_cmd_args.append(

command_table[alias_name])

alias_cmd = ServiceAliasCommand(*service_alias_cmd_args)

command_table[alias_name] = alias_cmd

class BaseAliasCommand(CLICommand):

_UNDOCUMENTED = True

def __init__(self, alias_name, alias_value):

"""Base class for alias command

:type alias_name: string

:param alias_name: The name of the alias

:type alias_value: string

:param alias_value: The parsed value of the alias. This can be

retrieved from `AliasLoader.get_aliases()[alias_name]`

"""

self._alias_name = alias_name

self._alias_value = alias_value

def __call__(self, args, parsed_args):

raise NotImplementedError('__call__')

@property

def name(self):

return self._alias_name

@name.setter

def name(self, value):

self._alias_name = value

class ServiceAliasCommand(BaseAliasCommand):

UNSUPPORTED_GLOBAL_PARAMETERS = [

'debug',

'profile'

]

def __init__(self, alias_name, alias_value, session, command_table,

parser, shadow_proxy_command=None):

"""Command for a `toplevel` subcommand alias

:type alias_name: string

:param alias_name: The name of the alias

:type alias_value: string

:param alias_value: The parsed value of the alias. This can be

retrieved from `AliasLoader.get_aliases()[alias_name]`

:type session: botocore.session.Session

:param session: The botocore session

:type command_table: dict

:param command_table: The command table containing all of the

possible service command objects that a particular alias could

redirect to.

:type parser: awscli.argparser.MainArgParser

:param parser: The parser to parse commands provided at the top level

of a CLI command which includes service commands and global

parameters. This is used to parse the service commmand and any

global parameters from the alias's value.

:type shadow_proxy_command: CLICommand

:param shadow_proxy_command: A built-in command that

potentially shadows the alias in name. If the alias

references this command in its value, the alias should proxy

to this command as oppposed to proxy to itself in the command

table

"""

super(ServiceAliasCommand, self).__init__(alias_name, alias_value)

self._session = session

self._command_table = command_table

self._parser = parser

self._shadow_proxy_command = shadow_proxy_command

def __call__(self, args, parsed_globals):

alias_args = self._get_alias_args()

parsed_alias_args, remaining = self._parser.parse_known_args(

alias_args)

self._update_parsed_globals(parsed_alias_args, parsed_globals)

# Take any of the remaining arguments that were not parsed out and

# prepend them to the remaining args provided to the alias.

remaining.extend(args)

LOG.debug(

'Alias %r passing on arguments: %r to %r command',

self._alias_name, remaining, parsed_alias_args.command)

# Pass the update remaing args and global args to the service command

# the alias proxied to.

command = self._command_table[parsed_alias_args.command]

if self._shadow_proxy_command:

shadow_name = self._shadow_proxy_command.name

# Use the shadow command only if the aliases value

# uses that command indicating it needs to proxy over to

# a built-in command.

if shadow_name == parsed_alias_args.command:

LOG.debug(

'Using shadowed command object: %s '

'for alias: %s', self._shadow_proxy_command,

self._alias_name

)

command = self._shadow_proxy_command

return command(remaining, parsed_globals)

def _get_alias_args(self):

try:

alias_args = shlex.split(self._alias_value)

except ValueError as e:

raise InvalidAliasException(

'Value of alias "%s" could not be parsed. '

'Received error: %s when parsing:\n%s' % (

self._alias_name, e, self._alias_value)

)

alias_args = [arg.strip(os.linesep) for arg in alias_args]

LOG.debug(

'Expanded subcommand alias %r with value: %r to: %r',

self._alias_name, self._alias_value, alias_args

)

return alias_args

def _update_parsed_globals(self, parsed_alias_args, parsed_globals):

global_params_to_update = self._get_global_parameters_to_update(

parsed_alias_args)

# Emit the top level args parsed event to ensure all possible

# customizations that typically get applied are applied to the

# global parameters provided in the alias before updating

# the original provided global parameter values

# and passing those onto subsequent commands.

emit_top_level_args_parsed_event(self._session, parsed_alias_args)

for param_name in global_params_to_update:

updated_param_value = getattr(parsed_alias_args, param_name)

setattr(parsed_globals, param_name, updated_param_value)

def _get_global_parameters_to_update(self, parsed_alias_args):

# Retrieve a list of global parameters that the newly parsed args

# from the alias will have to clobber from the originally provided

# parsed globals.

global_params_to_update = []

for parsed_param, value in vars(parsed_alias_args).items():

# To determine which parameters in the alias were global values

# compare the parsed alias parameters to the default as

# specified by the parser. If the parsed values from the alias

# differs from the default value in the parser,

# that global parameter must have been provided in the alias.

if self._parser.get_default(parsed_param) != value:

if parsed_param in self.UNSUPPORTED_GLOBAL_PARAMETERS:

raise InvalidAliasException(

'Global parameter "--%s" detected in alias "%s" '

'which is not support in subcommand aliases.' % (

parsed_param, self._alias_name))

else:

global_params_to_update.append(parsed_param)

return global_params_to_update

class ExternalAliasCommand(BaseAliasCommand):

def __init__(self, alias_name, alias_value, invoker=subprocess.call):

"""Command for external aliases

Executes command external of CLI as opposed to being a proxy

to another command.

:type alias_name: string

:param alias_name: The name of the alias

:type alias_value: string

:param alias_value: The parsed value of the alias. This can be

retrieved from `AliasLoader.get_aliases()[alias_name]`

:type invoker: callable

:param invoker: Callable to run arguments of external alias. The

signature should match that of ``subprocess.call``

"""

self._alias_name = alias_name

self._alias_value = alias_value

self._invoker = invoker

def __call__(self, args, parsed_globals):

command_components = [

self._alias_value[1:]

]

command_components.extend(args)

command = ' '.join(command_components)

LOG.debug(

'Using external alias %r with value: %r to run: %r',

self._alias_name, self._alias_value, command)

return self._invoker(command, shell=True)

aws-cli/awscli/alias.py at develop · rubythonode/aws-cli · GitHub

Expand file tree

Breadcrumbs

alias.py

Latest commit

History

Breadcrumbs

alias.py

File metadata and controls