/usr/share/pyshared/txaws/service.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 88 89 90 91 92 93 94 95 96 97 98 99 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 | # Copyright (C) 2009 Duncan McGreggor <duncan@canonical.com>
# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
# Licenced under the txaws licence available at /LICENSE in the txaws source.
from txaws.credentials import AWSCredentials
from txaws import regions
from txaws.util import parse
__all__ = ["AWSServiceEndpoint", "AWSServiceRegion", "REGION_US", "REGION_EU"]
# These old variable names are maintained for backwards compatibility.
REGION_US = regions.REGION_US
REGION_EU = regions.REGION_EU
EC2_ENDPOINT_US = regions.EC2_ENDPOINT_US
EC2_ENDPOINT_EU = regions.EC2_ENDPOINT_EU
S3_ENDPOINT = regions.S3_ENDPOINT
class AWSServiceEndpoint(object):
"""
@param uri: The URL for the service.
@param method: The HTTP method used when accessing a service.
@param ssl_hostname_verification: Whether or not SSL hotname verification
will be done when connecting to the endpoint.
"""
def __init__(self, uri="", method="GET", ssl_hostname_verification=False):
self.host = ""
self.port = None
self.path = "/"
self.method = method
self.ssl_hostname_verification = ssl_hostname_verification
self._parse_uri(uri)
if not self.scheme:
self.scheme = "http"
def _parse_uri(self, uri):
scheme, host, port, path = parse(
str(uri), defaultPort=False)
self.scheme = scheme
self.host = host
self.port = port
self.path = path
def set_host(self, host):
self.host = host
def get_host(self):
return self.host
def get_canonical_host(self):
"""
Return the canonical host as for the Host HTTP header specification.
"""
host = self.host.lower()
if self.port is not None:
host = "%s:%s" % (host, self.port)
return host
def set_canonical_host(self, canonical_host):
"""
Set host and port from a canonical host string as for the Host HTTP
header specification.
"""
parts = canonical_host.lower().split(":")
self.host = parts[0]
if len(parts) > 1 and parts[1]:
self.port = int(parts[1])
else:
self.port = None
def set_path(self, path):
self.path = path
def get_uri(self):
"""Get a URL representation of the service."""
uri = "%s://%s%s" % (self.scheme, self.get_canonical_host(), self.path)
return uri
def set_method(self, method):
self.method = method
class AWSServiceRegion(object):
"""
This object represents a collection of client factories that use the same
credentials. With Amazon, this collection is associated with a region
(e.g., US or EU).
@param creds: an AWSCredentials instance, optional.
@param access_key: The access key to use. This is only checked if no creds
parameter was passed.
@param secret_key: The secret key to use. This is only checked if no creds
parameter was passed.
@param region: a string value that represents the region that the
associated creds will be used against a collection of services.
@param uri: an endpoint URI that, if provided, will override the region
parameter.
@param method: The method argument forwarded to L{AWSServiceEndpoint}.
"""
# XXX update unit test to check for both ec2 and s3 endpoints
def __init__(self, creds=None, access_key="", secret_key="",
region=REGION_US, uri="", ec2_uri="", s3_uri="",
method="GET"):
if not creds:
creds = AWSCredentials(access_key, secret_key)
self.creds = creds
# Provide backwards compatibility for the "uri" parameter.
if uri and not ec2_uri:
ec2_uri = uri
if not ec2_uri and region == REGION_US:
ec2_uri = EC2_ENDPOINT_US
elif not ec2_uri and region == REGION_EU:
ec2_uri = EC2_ENDPOINT_EU
if not s3_uri:
s3_uri = S3_ENDPOINT
self._clients = {}
self.ec2_endpoint = AWSServiceEndpoint(uri=ec2_uri, method=method)
self.s3_endpoint = AWSServiceEndpoint(uri=s3_uri, method=method)
def get_client(self, cls, purge_cache=False, *args, **kwds):
"""
This is a general method for getting a client: if present, it is pulled
from the cache; if not, a new one is instantiated and then put into the
cache. This method should not be called directly, but rather by other
client-specific methods (e.g., get_ec2_client).
"""
key = str(cls) + str(args) + str(kwds)
instance = self._clients.get(key)
if purge_cache or not instance:
instance = cls(*args, **kwds)
self._clients[key] = instance
return instance
def get_ec2_client(self, creds=None):
from txaws.ec2.client import EC2Client
if creds:
self.creds = creds
return self.get_client(EC2Client, creds=self.creds,
endpoint=self.ec2_endpoint, query_factory=None)
def get_s3_client(self, creds=None):
from txaws.s3.client import S3Client
if creds:
self.creds = creds
return self.get_client(S3Client, creds=self.creds,
endpoint=self.s3_endpoint, query_factory=None)
|