def test_integrity_error(self):
"""
Tests the configuration test tool when an IntegrityError is raised.
"""
self.page.config_test_value = json.dumps({'text': 'test'})
with patch('django.forms.ModelForm.save',
side_effect=IntegrityError('foo')):
with LogCapture('cyphon.admin') as log_capture:
actual = self.page.run_test()
expected = "Could not create an object for testing: foo"
self.assertEqual(actual, expected)
msg = 'An error occurred while creating a test instance: ' + \
'<WSGIRequest: POST ' + \
"'/admin/datacondensers/datacondenser/1/change/test/'>"
log_capture.check(
('cyphon.admin', 'ERROR', msg),
)
python类IntegrityError()的实例源码
def test_integrity_error(self):
"""
Tests the configuration test tool when an IntegrityError is raised.
"""
with patch('django.forms.ModelForm.save',
side_effect=IntegrityError('foo')):
with LogCapture('cyphon.admin') as log_capture:
actual = self.page.run_test()
expected = "Could not create an object for testing: foo"
self.assertEqual(actual, expected)
msg = 'An error occurred while creating a test instance: ' + \
'<WSGIRequest: POST ' + \
"'/admin/logcondensers/logcondenser/1/change/test/'>"
log_capture.check(
('cyphon.admin', 'ERROR', msg),
)
def _commit(self):
if self.connection is not None:
try:
return self.connection.commit()
except Database.DatabaseError as e:
# cx_Oracle 5.0.4 raises a cx_Oracle.DatabaseError exception
# with the following attributes and values:
# code = 2091
# message = 'ORA-02091: transaction rolled back
# 'ORA-02291: integrity constraint (TEST_DJANGOTEST.SYS
# _C00102056) violated - parent key not found'
# We convert that particular case to our IntegrityError exception
x = e.args[0]
if hasattr(x, 'code') and hasattr(x, 'message') \
and x.code == 2091 and 'ORA-02291' in x.message:
six.reraise(utils.IntegrityError, utils.IntegrityError(*tuple(e.args)), sys.exc_info()[2])
raise
# Oracle doesn't support releasing savepoints. But we fake them when query
# logging is enabled to keep query counts consistent with other backends.
def executemany(self, query, params=None):
if not params:
# No params given, nothing to do
return None
# uniform treatment for sequences and iterables
params_iter = iter(params)
query, firstparams = self._fix_for_params(query, next(params_iter))
# we build a list of formatted params; as we're going to traverse it
# more than once, we can't make it lazy by using a generator
formatted = [firstparams] + [self._format_params(p) for p in params_iter]
self._guess_input_sizes(formatted)
try:
return self.cursor.executemany(query,
[self._param_generator(p) for p in formatted])
except Database.DatabaseError as e:
# cx_Oracle <= 4.4.0 wrongly raises a DatabaseError for ORA-01400.
if hasattr(e.args[0], 'code') and e.args[0].code == 1400 and not isinstance(e, IntegrityError):
six.reraise(utils.IntegrityError, utils.IntegrityError(*tuple(e.args)), sys.exc_info()[2])
raise
def handle(self, *args, **options):
with open(options['path'], 'r') as input_data:
data = csv.reader(input_data, delimiter='\t')
for row in data:
name, address, phone, lat, lng = row
logger.info(row)
try:
hospital = Hospital(
hospital_id=shortuuid.uuid(),
name=name,
address=address,
phone=phone,
opening_hours='',
lng=lng,
lat=lat
)
hospital.save(using=options['database'])
except IntegrityError:
print('data have already been imported')
break
def handle(self, *args, **options):
with open('dengue_linebot/data/tainan_minarea.json') as file:
data = json.load(file)
for area in data['features']:
try:
minarea = MinArea(
area_id=area['properties']['VILLAGEID'],
area_sn=area['properties']['VILLAGESN'],
area_name=area['properties']['VILLAGENAM'],
district_name=area['properties']['TOWNNAME'],
area=GEOSGeometry(json.dumps(area['geometry']))
)
minarea.save()
except IntegrityError:
self.stderr.write('data have already been imported')
break
self.stdout.write('Successfully imported')
def _rename(self, apps, schema_editor, old_model, new_model):
ContentType = apps.get_model('contenttypes', 'ContentType')
db = schema_editor.connection.alias
if not router.allow_migrate_model(db, ContentType):
return
try:
content_type = ContentType.objects.db_manager(db).get_by_natural_key(self.app_label, old_model)
except ContentType.DoesNotExist:
pass
else:
content_type.model = new_model
try:
with transaction.atomic(using=db):
content_type.save(update_fields={'model'})
except IntegrityError:
# Gracefully fallback if a stale content type causes a
# conflict as update_contenttypes will take care of asking the
# user what should be done next.
content_type.model = old_model
else:
# Clear the cache as the `get_by_natual_key()` call will cache
# the renamed ContentType instance by its old model name.
ContentType.objects.clear_cache()
def _rename(self, apps, schema_editor, old_model, new_model):
ContentType = apps.get_model('contenttypes', 'ContentType')
db = schema_editor.connection.alias
if not router.allow_migrate_model(db, ContentType):
return
try:
content_type = ContentType.objects.db_manager(db).get_by_natural_key(self.app_label, old_model)
except ContentType.DoesNotExist:
pass
else:
content_type.model = new_model
try:
with transaction.atomic(using=db):
content_type.save(update_fields={'model'})
except IntegrityError:
# Gracefully fallback if a stale content type causes a
# conflict as update_contenttypes will take care of asking the
# user what should be done next.
content_type.model = old_model
else:
# Clear the cache as the `get_by_natual_key()` call will cache
# the renamed ContentType instance by its old model name.
ContentType.objects.clear_cache()
def _rename(self, apps, schema_editor, old_model, new_model):
ContentType = apps.get_model('contenttypes', 'ContentType')
db = schema_editor.connection.alias
if not router.allow_migrate_model(db, ContentType):
return
try:
content_type = ContentType.objects.db_manager(db).get_by_natural_key(self.app_label, old_model)
except ContentType.DoesNotExist:
pass
else:
content_type.model = new_model
try:
with transaction.atomic(using=db):
content_type.save(update_fields={'model'})
except IntegrityError:
# Gracefully fallback if a stale content type causes a
# conflict as remove_stale_contenttypes will take care of
# asking the user what should be done next.
content_type.model = old_model
else:
# Clear the cache as the `get_by_natual_key()` call will cache
# the renamed ContentType instance by its old model name.
ContentType.objects.clear_cache()
def mig_users(self, cnx, cursor):
query = 'SELECT * FROM user;'
cursor.execute(query)
o2n_map = {}
for user in cursor:
new_user = Poster(username=user['username'],
email=user['email'],
password=user['token'],
backend='ISS.auth.backends.vB5_%s' % user['scheme'])
try:
new_user.save()
o2n_map[user['userid']] = new_user.pk
except IntegrityError:
o2n_map[user['userid']] = Poster.objects.get(
username=user['username']).pk
return o2n_map
def mig_thanks(self, cnx, cursor, user_pk_map, post_pk_map):
cursor.execute("""
SELECT contentid, userid, receiveduserid FROM dbtech_thanks_entry
WHERE varname='thanks'
GROUP BY contentid, userid, receiveduserid
""")
for thanks in cursor:
try:
post_id = post_pk_map[thanks['contentid']]
except KeyError:
print ('Error migrating thanks: post %d was not found'
% thanks['contentid'])
else:
new_thanks = Thanks(
thanker_id=user_pk_map[thanks['userid']],
thankee_id=user_pk_map[thanks['receiveduserid']],
post_id=post_id)
try:
new_thanks.save()
except IntegrityError, e:
import pdb; pdb.set_trace()
return
def _commit(self):
if self.connection is not None:
try:
return self.connection.commit()
except Database.DatabaseError as e:
# cx_Oracle 5.0.4 raises a cx_Oracle.DatabaseError exception
# with the following attributes and values:
# code = 2091
# message = 'ORA-02091: transaction rolled back
# 'ORA-02291: integrity constraint (TEST_DJANGOTEST.SYS
# _C00102056) violated - parent key not found'
# We convert that particular case to our IntegrityError exception
x = e.args[0]
if hasattr(x, 'code') and hasattr(x, 'message') \
and x.code == 2091 and 'ORA-02291' in x.message:
six.reraise(utils.IntegrityError, utils.IntegrityError(*tuple(e.args)), sys.exc_info()[2])
raise
# Oracle doesn't support releasing savepoints. But we fake them when query
# logging is enabled to keep query counts consistent with other backends.
def executemany(self, query, params=None):
if not params:
# No params given, nothing to do
return None
# uniform treatment for sequences and iterables
params_iter = iter(params)
query, firstparams = self._fix_for_params(query, next(params_iter))
# we build a list of formatted params; as we're going to traverse it
# more than once, we can't make it lazy by using a generator
formatted = [firstparams] + [self._format_params(p) for p in params_iter]
self._guess_input_sizes(formatted)
try:
return self.cursor.executemany(query,
[self._param_generator(p) for p in formatted])
except Database.DatabaseError as e:
# cx_Oracle <= 4.4.0 wrongly raises a DatabaseError for ORA-01400.
if hasattr(e.args[0], 'code') and e.args[0].code == 1400 and not isinstance(e, IntegrityError):
six.reraise(utils.IntegrityError, utils.IntegrityError(*tuple(e.args)), sys.exc_info()[2])
raise
def _rename(self, apps, schema_editor, old_model, new_model):
ContentType = apps.get_model('contenttypes', 'ContentType')
db = schema_editor.connection.alias
if not router.allow_migrate_model(db, ContentType):
return
try:
content_type = ContentType.objects.db_manager(db).get_by_natural_key(self.app_label, old_model)
except ContentType.DoesNotExist:
pass
else:
content_type.model = new_model
try:
with transaction.atomic(using=db):
content_type.save(update_fields={'model'})
except IntegrityError:
# Gracefully fallback if a stale content type causes a
# conflict as remove_stale_contenttypes will take care of
# asking the user what should be done next.
content_type.model = old_model
else:
# Clear the cache as the `get_by_natual_key()` call will cache
# the renamed ContentType instance by its old model name.
ContentType.objects.clear_cache()
def _commit(self):
if self.connection is not None:
try:
return self.connection.commit()
except Database.DatabaseError as e:
# cx_Oracle raises a cx_Oracle.DatabaseError exception
# with the following attributes and values:
# code = 2091
# message = 'ORA-02091: transaction rolled back
# 'ORA-02291: integrity constraint (TEST_DJANGOTEST.SYS
# _C00102056) violated - parent key not found'
# We convert that particular case to our IntegrityError exception
x = e.args[0]
if hasattr(x, 'code') and hasattr(x, 'message') \
and x.code == 2091 and 'ORA-02291' in x.message:
six.reraise(utils.IntegrityError, utils.IntegrityError(*tuple(e.args)), sys.exc_info()[2])
raise
# Oracle doesn't support releasing savepoints. But we fake them when query
# logging is enabled to keep query counts consistent with other backends.
