def _connect(self):
args = dict(aws_access_key_id=self.db_user,
aws_secret_access_key=self.db_passwd,
is_secure=self.enable_ssl)
try:
region = [x for x in boto.sdb.regions() if x.endpoint == self.db_host][0]
args['region'] = region
except IndexError:
pass
self._sdb = boto.connect_sdb(**args)
# This assumes that the domain has already been created
# It's much more efficient to do it this way rather than
# having this make a roundtrip each time to validate.
# The downside is that if the domain doesn't exist, it breaks
self._domain = self._sdb.lookup(self.db_name, validate=False)
if not self._domain:
self._domain = self._sdb.create_domain(self.db_name)
python类connect_sdb()的实例源码
def load_from_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
item = domain.get_item(item_name)
for section in item.keys():
if not self.has_section(section):
self.add_section(section)
d = json.loads(item[section])
for attr_name in d.keys():
attr_value = d[attr_name]
if attr_value is None:
attr_value = 'None'
if isinstance(attr_value, bool):
self.setbool(section, attr_name, attr_value)
else:
self.set(section, attr_name, attr_value)
def _connect(self):
args = dict(aws_access_key_id=self.db_user,
aws_secret_access_key=self.db_passwd,
is_secure=self.enable_ssl)
try:
region = [x for x in boto.sdb.regions() if x.endpoint == self.db_host][0]
args['region'] = region
except IndexError:
pass
self._sdb = boto.connect_sdb(**args)
# This assumes that the domain has already been created
# It's much more efficient to do it this way rather than
# having this make a roundtrip each time to validate.
# The downside is that if the domain doesn't exist, it breaks
self._domain = self._sdb.lookup(self.db_name, validate=False)
if not self._domain:
self._domain = self._sdb.create_domain(self.db_name)
def load_from_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
item = domain.get_item(item_name)
for section in item.keys():
if not self.has_section(section):
self.add_section(section)
d = json.loads(item[section])
for attr_name in d.keys():
attr_value = d[attr_name]
if attr_value == None:
attr_value = 'None'
if isinstance(attr_value, bool):
self.setbool(section, attr_name, attr_value)
else:
self.set(section, attr_name, attr_value)
def load_from_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
item = domain.get_item(item_name)
for section in item.keys():
if not self.has_section(section):
self.add_section(section)
d = json.loads(item[section])
for attr_name in d.keys():
attr_value = d[attr_name]
if attr_value == None:
attr_value = 'None'
if isinstance(attr_value, bool):
self.setbool(section, attr_name, attr_value)
else:
self.set(section, attr_name, attr_value)
def _connect(self):
args = dict(aws_access_key_id=self.db_user,
aws_secret_access_key=self.db_passwd,
is_secure=self.enable_ssl)
try:
region = [x for x in boto.sdb.regions() if x.endpoint == self.db_host][0]
args['region'] = region
except IndexError:
pass
self._sdb = boto.connect_sdb(**args)
# This assumes that the domain has already been created
# It's much more efficient to do it this way rather than
# having this make a roundtrip each time to validate.
# The downside is that if the domain doesn't exist, it breaks
self._domain = self._sdb.lookup(self.db_name, validate=False)
if not self._domain:
self._domain = self._sdb.create_domain(self.db_name)
def load_from_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
item = domain.get_item(item_name)
for section in item.keys():
if not self.has_section(section):
self.add_section(section)
d = json.loads(item[section])
for attr_name in d.keys():
attr_value = d[attr_name]
if attr_value is None:
attr_value = 'None'
if isinstance(attr_value, bool):
self.setbool(section, attr_name, attr_value)
else:
self.set(section, attr_name, attr_value)
def load_from_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
item = domain.get_item(item_name)
for section in item.keys():
if not self.has_section(section):
self.add_section(section)
d = json.loads(item[section])
for attr_name in d.keys():
attr_value = d[attr_name]
if attr_value == None:
attr_value = 'None'
if isinstance(attr_value, bool):
self.setbool(section, attr_name, attr_value)
else:
self.set(section, attr_name, attr_value)
def _connect(self):
"""Connect to our domain"""
if not self._db:
import boto
sdb = boto.connect_sdb()
if not self.domain_name:
self.domain_name = boto.config.get("DB", "sequence_db", boto.config.get("DB", "db_name", "default"))
try:
self._db = sdb.get_domain(self.domain_name)
except SDBResponseError as e:
if e.status == 400:
self._db = sdb.create_domain(self.domain_name)
else:
raise
return self._db
def dump_to_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
if not domain:
domain = sdb.create_domain(domain_name)
item = domain.new_item(item_name)
item.active = False
for section in self.sections():
d = {}
for option in self.options(section):
d[option] = self.get(section, option)
item[section] = json.dumps(d)
item.save()
def _connect(self):
"""Connect to our domain"""
if not self._db:
import boto
sdb = boto.connect_sdb()
if not self.domain_name:
self.domain_name = boto.config.get("DB", "sequence_db", boto.config.get("DB", "db_name", "default"))
try:
self._db = sdb.get_domain(self.domain_name)
except SDBResponseError as e:
if e.status == 400:
self._db = sdb.create_domain(self.domain_name)
else:
raise
return self._db
def dump_to_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
if not domain:
domain = sdb.create_domain(domain_name)
item = domain.new_item(item_name)
item.active = False
for section in self.sections():
d = {}
for option in self.options(section):
d[option] = self.get(section, option)
item[section] = json.dumps(d)
item.save()
def dump_to_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
if not domain:
domain = sdb.create_domain(domain_name)
item = domain.new_item(item_name)
item.active = False
for section in self.sections():
d = {}
for option in self.options(section):
d[option] = self.get(section, option)
item[section] = json.dumps(d)
item.save()
def _connect(self):
"""Connect to our domain"""
if not self._db:
import boto
sdb = boto.connect_sdb()
if not self.domain_name:
self.domain_name = boto.config.get("DB", "sequence_db", boto.config.get("DB", "db_name", "default"))
try:
self._db = sdb.get_domain(self.domain_name)
except SDBResponseError as e:
if e.status == 400:
self._db = sdb.create_domain(self.domain_name)
else:
raise
return self._db
def dump_to_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
if not domain:
domain = sdb.create_domain(domain_name)
item = domain.new_item(item_name)
item.active = False
for section in self.sections():
d = {}
for option in self.options(section):
d[option] = self.get(section, option)
item[section] = json.dumps(d)
item.save()
def dump_to_sdb(self, domain_name, item_name):
from boto.compat import json
sdb = boto.connect_sdb()
domain = sdb.lookup(domain_name)
if not domain:
domain = sdb.create_domain(domain_name)
item = domain.new_item(item_name)
item.active = False
for section in self.sections():
d = {}
for option in self.options(section):
d[option] = self.get(section, option)
item[section] = json.dumps(d)
item.save()
def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,
is_secure=True, port=None, proxy=None, proxy_port=None,
proxy_user=None, proxy_pass=None, debug=0,
https_connection_factory=None, region=None, path='/',
converter=None, security_token=None, validate_certs=True,
profile_name=None):
"""
For any keywords that aren't documented, refer to the parent class,
:py:class:`boto.connection.AWSAuthConnection`. You can avoid having
to worry about these keyword arguments by instantiating these objects
via :py:func:`boto.connect_sdb`.
:type region: :class:`boto.sdb.regioninfo.SDBRegionInfo`
:keyword region: Explicitly specify a region. Defaults to ``us-east-1``
if not specified. You may also specify the region in your ``boto.cfg``:
.. code-block:: cfg
[SDB]
region = eu-west-1
"""
if not region:
region_name = boto.config.get('SDB', 'region', self.DefaultRegionName)
for reg in boto.sdb.regions():
if reg.name == region_name:
region = reg
break
self.region = region
super(SDBConnection, self).__init__(aws_access_key_id,
aws_secret_access_key,
is_secure, port, proxy,
proxy_port, proxy_user, proxy_pass,
self.region.endpoint, debug,
https_connection_factory, path,
security_token=security_token,
validate_certs=validate_certs,
profile_name=profile_name)
self.box_usage = 0.0
self.converter = converter
self.item_cls = Item
def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,
is_secure=True, port=None, proxy=None, proxy_port=None,
proxy_user=None, proxy_pass=None, debug=0,
https_connection_factory=None, region=None, path='/',
converter=None, security_token=None, validate_certs=True,
profile_name=None):
"""
For any keywords that aren't documented, refer to the parent class,
:py:class:`boto.connection.AWSAuthConnection`. You can avoid having
to worry about these keyword arguments by instantiating these objects
via :py:func:`boto.connect_sdb`.
:type region: :class:`boto.sdb.regioninfo.SDBRegionInfo`
:keyword region: Explicitly specify a region. Defaults to ``us-east-1``
if not specified. You may also specify the region in your ``boto.cfg``:
.. code-block:: cfg
[SDB]
region = eu-west-1
"""
if not region:
region_name = boto.config.get('SDB', 'region', self.DefaultRegionName)
for reg in boto.sdb.regions():
if reg.name == region_name:
region = reg
break
self.region = region
super(SDBConnection, self).__init__(aws_access_key_id,
aws_secret_access_key,
is_secure, port, proxy,
proxy_port, proxy_user, proxy_pass,
self.region.endpoint, debug,
https_connection_factory, path,
security_token=security_token,
validate_certs=validate_certs,
profile_name=profile_name)
self.box_usage = 0.0
self.converter = converter
self.item_cls = Item
def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,
is_secure=True, port=None, proxy=None, proxy_port=None,
proxy_user=None, proxy_pass=None, debug=0,
https_connection_factory=None, region=None, path='/',
converter=None, security_token=None, validate_certs=True,
profile_name=None):
"""
For any keywords that aren't documented, refer to the parent class,
:py:class:`boto.connection.AWSAuthConnection`. You can avoid having
to worry about these keyword arguments by instantiating these objects
via :py:func:`boto.connect_sdb`.
:type region: :class:`boto.sdb.regioninfo.SDBRegionInfo`
:keyword region: Explicitly specify a region. Defaults to ``us-east-1``
if not specified. You may also specify the region in your ``boto.cfg``:
.. code-block:: cfg
[SDB]
region = eu-west-1
"""
if not region:
region_name = boto.config.get('SDB', 'region', self.DefaultRegionName)
for reg in boto.sdb.regions():
if reg.name == region_name:
region = reg
break
self.region = region
super(SDBConnection, self).__init__(aws_access_key_id,
aws_secret_access_key,
is_secure, port, proxy,
proxy_port, proxy_user, proxy_pass,
self.region.endpoint, debug,
https_connection_factory, path,
security_token=security_token,
validate_certs=validate_certs,
profile_name=profile_name)
self.box_usage = 0.0
self.converter = converter
self.item_cls = Item
