openstack
/
panko
Code Issues Proposed changes

sqlalchemy: use nested transaction when getting/creating event types 

If adding the database fails, not using a nested transaction will make the
whole transaction passed from the caller fail. The code does not handle that at
all.

This switches to using a nested transaction, so only the insert is rolled-back
if it fails.

Change-Id: I5196147524e5fdd0d46d4c1995a8afb964ce3f6f
(cherry picked from commit e4021dbeac)
This commit is contained in:
Julien Danjou 2017-03-22 13:09:29 +01:00
parent 9a9ae04a86
commit efacfce94c
2 changed files with 16 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
panko/storage/impl_sqlalchemy.py
View File

@ -146,15 +146,13 @@ class Connection(base.Connection):
engine.execute(table.delete())
engine.dispose()
def _get_or_create_event_type(self, event_type, session=None):
def _get_or_create_event_type(self, event_type, session):
"""Check if an event type with the supplied name is already exists.
If not, we create it and return the record. This may result in a flush.
"""
try:
if session is None:
session = self._engine_facade.get_session()
with session.begin(subtransactions=True):
with session.begin(nested=True):
et = session.query(models.EventType).filter(
models.EventType.desc == event_type).first()
if not et:

18
panko/tests/functional/storage/test_impl_sqlalchemy.py
View File

@ -43,22 +43,32 @@ class EventTypeTest(tests_db.TestBase):
# EventType is a construct specific to sqlalchemy
# Not applicable to other drivers.
def setUp(self):
super(EventTypeTest, self).setUp()
self.session = self.conn._engine_facade.get_session()
self.session.begin()
def test_event_type_exists(self):
et1 = self.conn._get_or_create_event_type("foo")
et1 = self.conn._get_or_create_event_type("foo", self.session)
self.assertTrue(et1.id >= 0)
et2 = self.conn._get_or_create_event_type("foo")
et2 = self.conn._get_or_create_event_type("foo", self.session)
self.assertEqual(et2.id, et1.id)
self.assertEqual(et2.desc, et1.desc)
def test_event_type_unique(self):
et1 = self.conn._get_or_create_event_type("foo")
et1 = self.conn._get_or_create_event_type("foo", self.session)
self.assertTrue(et1.id >= 0)
et2 = self.conn._get_or_create_event_type("blah")
et2 = self.conn._get_or_create_event_type("blah", self.session)
self.assertNotEqual(et1.id, et2.id)
self.assertNotEqual(et1.desc, et2.desc)
# Test the method __repr__ returns a string
self.assertTrue(reprlib.repr(et2))
def tearDown(self):
self.session.rollback()
self.session.close()
super(EventTypeTest, self).tearDown()
@tests_db.run_with('sqlite', 'mysql', 'pgsql')
class EventTest(tests_db.TestBase):