def get_summoners(self, match):
self.delay()
participants = self.watcher.get_match(match)['participantIdentities']
for participant in participants:
summoner = Summoner(participant['player']['summonerId'])
self.session.add(summoner)
try:
self.session.commit()
print('New Summoner - %s' % summoner)
except IntegrityError:
self.session.rollback()
print('Duplicate Summoner - %s' % summoner)
except FlushError:
self.session.rollback()
print('FlushError Summoner - %s' % summoner)
python类FlushError()的实例源码
def __bulk_add_new_file_dids(files, account, dataset_meta=None, session=None):
"""
Bulk add new dids.
:param dids: the list of new files.
:param account: The account owner.
:param session: The database session in use.
:returns: True is successful.
"""
for file in files:
new_did = models.DataIdentifier(scope=file['scope'], name=file['name'],
account=file.get('account') or account,
did_type=DIDType.FILE, bytes=file['bytes'],
md5=file.get('md5'), adler32=file.get('adler32'),
is_new=None)
for key in file.get('meta', []):
new_did.update({key: file['meta'][key]})
for key in dataset_meta or {}:
new_did.update({key: dataset_meta[key]})
new_did.save(session=session, flush=False)
try:
session.flush()
except IntegrityError, error:
raise exception.RucioException(error.args)
except DatabaseError, error:
raise exception.RucioException(error.args)
except FlushError, error:
if match('New instance .* with identity key .* conflicts with persistent instance', error.args[0]):
raise exception.DataIdentifierAlreadyExists('Data Identifier already exists!')
raise exception.RucioException(error.args)
return True
def test_add_regions_errors(self, mock_gfbu, mock_strin):
ret_flavor = MagicMock()
ret_flavor.flavor.regions = [Region(name='test_region')]
mock_gfbu.return_value = ret_flavor
global error
error = 1
injector.override_injected_dependency(('data_manager', get_datamanager_mock))
self.assertRaises(flavor_logic.ErrorStatus, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region')]),
'transaction')
error = 4
injector.override_injected_dependency(
('data_manager', get_datamanager_mock))
mock_strin.side_effect = exc.FlushError()
self.assertRaises(exc.FlushError, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region')]),
'transaction')
mock_strin.side_effect = exc.FlushError(
'conflicts with persistent instance')
self.assertRaises(flavor_logic.ErrorStatus, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region')]),
'transaction')
mock_strin.side_effect = ValueError()
self.assertRaises(ValueError, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region')]),
'transaction')
mock_strin.side_effect = ValueError(
'conflicts with persistent instance')
self.assertRaises(flavor_logic.ConflictError, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region')]),
'transaction')
self.assertRaises(flavor_logic.ErrorStatus, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='')]),
'transaction')
self.assertRaises(flavor_logic.ErrorStatus, flavor_logic.add_regions,
'uuid', RegionWrapper([Region(name='test_region', type='group')]),
'transaction')
def get_teams(self, summoners):
self.delay()
error = True
while error:
try:
teams = self.watcher.get_teams_for_summoners(summoners)
error = False
except LoLException:
pass
for key, value in teams.items():
for item in value:
try:
# Check if the team is 5x5.
if len(item['roster']['memberList']) < 5:
continue
except KeyError:
# LoL api is stupid and sometimes forgets to put a key
# in it's api return when it should put null.
continue
team = Team(item['fullId'])
self.session.add(team)
# Add all members in the 5x5 team to the summoners table.
for member in item['roster']['memberList']:
roster_member = TeamRosterMember(team.id, member['playerId'])
self.session.add(roster_member)
try:
self.session.commit()
print('New Team - %s' % team)
except IntegrityError:
self.session.rollback()
print('Duplicate Team - %s' % team)
continue
except FlushError:
self.session.rollback()
print('FlushError Team - %s' % team)
def test_keys(session):
from sqlalchemy.orm.exc import FlushError
from snovault.storage import (
Resource,
Key,
)
name = 'testdata'
props1 = {'foo': 'bar'}
resource = Resource('test_item', {name: props1})
session.add(resource)
session.flush()
resource = session.query(Resource).one()
testname = 'foo'
key = Key(rid=resource.rid, name=testname, value=props1[testname])
session.add(key)
session.flush()
assert session.query(Key).count() == 1
othertest = 'foofoo'
othervalue = 'barbar'
key2 = Key(rid=resource.rid, name=othertest, value=othervalue)
session.add(key2)
session.flush()
assert session.query(Key).count() == 2
props2 = {'foofoo': 'barbar'}
resource2 = Resource('test_item', {name: props2})
session.add(resource2)
session.flush()
key3 = Key(rid=resource2.rid, name=testname, value=props1[testname])
session.add(key3)
with pytest.raises(FlushError):
session.flush()
def calculate_mutation_strength(run_id, generation, mutation_strength_bin):
"""Query mutation_strength for bin and adjust as necessary.
Args:
run_id (str): identification string for run.
generation (int): iteration in bin-mutate-simulate routine.
parent (sqlalchemy.orm.query.Query): parent-material corresponding to
the bin being queried.
Returns:
mutation_strength.strength (float): mutation strength to be used for
parents in the bin being queried. If the fraction of children from
previous generation which populate the SAME bin as their parent is LESS
THAN 10% then the mutation strength is REDUCED BY 5%. If the fraction
of these children populating the SAME bin as their parent is GREATER
THAN 50% then the mutation strength is INCREASED BY 5%.
"""
mutation_strength_key = [run_id, generation] + mutation_strength_bin
mutation_strength = session.query(MutationStrength).get(mutation_strength_key)
if mutation_strength:
print("Mutation strength already calculated for this bin and generation.")
else:
print("Calculating mutation strength...")
mutation_strength = MutationStrength.get_prior(*mutation_strength_key).clone()
if config['interactive_mode'] == 'on':
print('Prior mutation strength :\t{}'.format(mutation_strength.strength))
mutation_strength.generation = generation
try:
fraction_in_parent_bin = calculate_percent_children_in_bin(run_id, generation, mutation_strength_bin)
if config['interactive_mode'] != 'on':
if fraction_in_parent_bin < 0.1 and mutation_strength.strength - 0.05 > 0:
mutation_strength.strength -= 0.05
elif fraction_in_parent_bin > 0.5 and mutation_strength.strength + 0.05 < 1:
mutation_strength.strength += 0.05
elif config['interactive_mode'] == 'on':
print('\tFor adaptive mutation strength(s), DECREASE strength \n' +
'\tby 5% if the fraction of children in parent-bin is \n' +
'\tLESS THAN 0.1; INCREASE strength by 5% if the \n' +
'\tfraction of children in parent-bin is GREATER THAN \n' +
'\t0.5.')
mutation_strength.strength = set_variable(
'Please Enter new mutation strength :\t', 'mutation_strength.strength')
except ZeroDivisionError:
print("No prior generation materials in this bin with children.")
try:
session.add(mutation_strength)
session.commit()
except (FlushError, IntegrityError) as e:
print("Somebody beat us to saving a row with this generation. That's ok!")
session.rollback()
# it's ok b/c this calculation should always yield the exact same result!
sys.stdout.flush()
return mutation_strength.strength
def test_add_tenants_errors(self, mock_gfbu, mock_strin):
ret_flavor = MagicMock()
tenants = ['test_tenant']
ret_flavor.flavor.tenants = tenants
mock_gfbu.return_value = ret_flavor
global error
error = 1
injector.override_injected_dependency(
('data_manager', get_datamanager_mock))
self.assertRaises(flavor_logic.ErrorStatus,
flavor_logic.add_tenants, 'uuid',
TenantWrapper(tenants),
'transaction')
# Flavor is public
error = 5
self.assertRaises(flavor_logic.ErrorStatus,
flavor_logic.add_tenants, 'uuid',
TenantWrapper(tenants),
'transaction')
error = 31
moq = MagicMock()
moq.tenants = [1337]
self.assertRaises(ValueError,
flavor_logic.add_tenants, 'uuid',
moq,
'transaction')
mock_strin.side_effect = exc.FlushError(
'conflicts with persistent instance')
self.assertRaises(flavor_logic.ConflictError,
flavor_logic.add_tenants, 'uuid',
TenantWrapper(tenants),
'transaction')
mock_strin.side_effect = exc.FlushError('')
self.assertRaises(exc.FlushError,
flavor_logic.add_tenants, 'uuid',
TenantWrapper(tenants),
'transaction')
