secret_key='aws_secret_access_key',
)
kwargs = dict((mapping[k],v) for (k,v) in conf.iteritems() if k in mapping)
- conn = boto.s3.connection.S3Connection(
- # TODO support & test all variations
- calling_format=boto.s3.connection.OrdinaryCallingFormat(),
- **kwargs
+ #process calling_format argument
+ calling_formats = dict(
+ ordinary=boto.s3.connection.OrdinaryCallingFormat(),
+ subdomain=boto.s3.connection.SubdomainCallingFormat(),
+ vhost=boto.s3.connection.VHostCallingFormat(),
)
+ kwargs['calling_format'] = calling_formats['ordinary']
+ if conf.has_key('calling_format'):
+ raw_calling_format = conf['calling_format']
+ try:
+ kwargs['calling_format'] = calling_formats[raw_calling_format]
+ except KeyError:
+ raise RuntimeError(
+ 'calling_format unknown: %r' % raw_calling_format
+ )
+ # TODO test vhost calling format
+ conn = boto.s3.connection.S3Connection(**kwargs)
return conn
def setup():
s3.clear()
config.clear()
+ calling_formats = dict(
+ ordinary=boto.s3.connection.OrdinaryCallingFormat(),
+ subdomain=boto.s3.connection.SubdomainCallingFormat(),
+ vhost=boto.s3.connection.VHostCallingFormat(),
+ )
for section in cfg.sections():
try:
(type_, name) = section.split(None, 1)
except ConfigParser.NoOptionError:
port = None
+ try:
+ raw_calling_format = cfg.get(section, 'calling_format')
+ except ConfigParser.NoOptionError:
+ raw_calling_format = 'ordinary'
+
+ try:
+ calling_format = calling_formats[raw_calling_format]
+ except KeyError:
+ raise RuntimeError(
+ 'calling_format unknown: %r' % raw_calling_format
+ )
+
config[name] = bunch.Bunch()
for var in [
'user_id',
is_secure=cfg.getboolean(section, 'is_secure'),
port=port,
host=cfg.get(section, 'host'),
- # TODO support & test all variations
- calling_format=boto.s3.connection.OrdinaryCallingFormat(),
+ # TODO test vhost calling format
+ calling_format=calling_format,
)
s3[name] = conn