/usr/share/pyshared/txaws/client/discover/command.py is in python-txaws 0.2.3-1ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | # Copyright (C) 2010 Jamu Kakar <jkakar@kakar.ca>
# Licenced under the txaws licence available at /LICENSE in the txaws source.
"""
A L{Command} object makes an arbitrary EC2 API method call and displays the
response received from the backend cloud.
"""
import sys
from txaws.ec2.client import Query
from txaws.exception import AWSError
from txaws.service import AWSServiceRegion
class Command(object):
"""
An EC2 API method call command that can make a request and display the
response received from the backend cloud.
@param key: The AWS access key ID to use when making the method call.
@param secret: The AWS secret key to sign the method call with.
@param endpoint: The URL of the cloud to invoke the method on.
@param parameters: A C{dict} with parameters to include with the method
call.
@param output: Optionally, a stream to write output to. Defaults to
C{sys.stdout}.
@param query_factory: Optionally, a factory to create the L{Query} object
used to invoke the method. Defaults to returning a L{Query} instance.
"""
def __init__(self, key, secret, endpoint, action, parameters, output=None,
query_factory=None):
self.key = key
self.secret = secret
self.endpoint = endpoint
self.action = action
self.parameters = parameters
if output is None:
output = sys.stdout
self.output = output
if query_factory is None:
query_factory = Query
self.query_factory = query_factory
def run(self):
"""
Run the configured method and write the HTTP response status and text
to the output stream.
"""
region = AWSServiceRegion(access_key=self.key, secret_key=self.secret,
uri=self.endpoint)
query = self.query_factory(action=self.action, creds=region.creds,
endpoint=region.ec2_endpoint,
other_params=self.parameters)
def write_response(response):
print >> self.output, "URL: %s" % query.client.url
print >> self.output
print >> self.output, "HTTP status code: %s" % query.client.status
print >> self.output
print >> self.output, response
def write_error(failure):
if failure.check(AWSError):
message = failure.value.original
else:
message = failure.getErrorMessage()
if message.startswith("Error Message: "):
message = message[len("Error Message: "):]
print >> self.output, "URL: %s" % query.client.url
print >> self.output
if getattr(query.client, "status", None) is not None:
print >> self.output, "HTTP status code: %s" % (
query.client.status,)
print >> self.output
print >> self.output, message
if getattr(failure.value, "response", None) is not None:
print >> self.output
print >> self.output, failure.value.response
deferred = query.submit()
deferred.addCallback(write_response)
deferred.addErrback(write_error)
return deferred
|