The Problem

Calling SQLAlchemy's session.rollback() after hitting a DB integrity error (requirement of SQLAlchemy) causes all session objects to be disposed. This includes objects created by previous selects.

Looking at the SQL generated our query data is caught in an implicit transaction that isn't finished until after we do session.close.

Simplified SQL Example

Begin (implicit) transaction
SELECT Data
# Should be end transaction here
# Should Start new Transaction
INSERT New Data
# hit integrity error
rollback

Problem Section

try:
    self.session.add(insertable['cluster'])
    self.session.commit()
except IntegrityError:
    print('Prevented from inserting duplicate to cluster table')
    self.session.rollback()
# insertable is now an empty object and can't be used without repopulating

What this means is that in the worst case we'll have an integrity error on every insert and we'll need to re-select the row IDs for each insert.

session.commit After Select

def find_entry(self, inpath):
    inpath = inpath.rstrip('/')
    inpath = '/'.join(inpath.split('/')[:-1])
    entry = self.session.query(
        models.Gather).filter_by(path=inpath).first()
    self.session.commit()
    return entry

But this doesn't actually commit the session transaction and so when we rollback we still lose the data.

Reasons And Other Effects

For performance reasons we have a set of code that selects data from a database, gets row IDs from two tables in said DB, and then uses the selected row IDs as foreign keys when we insert.

This is mainly done so that we don't have to query on each insert.

The problem is that if we run into a constraint or integrity errors while we're inserting than we have to do a session.rollback. We're finding that this session.rollback is killing our prior query even though they should logically be on different transactions.

In addition to our selected data if we have objects that inserted successfully that we'd like to reference for their IDs these are also removed after a sessions.rollback.

Full Code

class DBInserter:

    def __init__(self):
        # connection info here
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

    def __del__(self):
        self.session.close()
        self.engine.dispose()

    def find_entry(self, inpath):
        inpath = inpath.rstrip('/')
        inpath = '/'.join(inpath.split('/')[:-1])
        entry = self.session.query(
            models.Gather).filter_by(path=inpath).first()
        return entry

    def build_insertable(self, jsonin, gather, lnn):
        """
        Build sqlalchemy object that is ready to be inserted into the db
        """
        object_dict = {}

        cluster = models.Cluster(
            guid = jsonin.get('guid'),
            )

        gather_id = None
        if gather:
            gather_id = gather.gather_id
            cluster.gather_id = gather_id
            gather.cluster_guid = jsonin.get('guid')
            gather.cluster_name = jsonin.get('name'),
            object_dict['gather'] = gather

        node_gather = models.NodeGather(
            gather_id = gather_id,
            lnn = lnn,
            checksum = jsonin.get('checksum'),
            checksum_valid = jsonin.get('checksum_valid'),
            compliance = jsonin.get('compliance'),
            encoding = jsonin.get('encoding'),
            joinmode = jsonin.get('joinmode'),
            master = jsonin.get('master'),
            maxid = jsonin.get('maxid'),
            timezone = jsonin.get('timezone'),
            )

        object_dict['cluster'] = cluster
        object_dict['node_gather'] = node_gather
        return object_dict

    def insert(self, insertable):
        """
        insert prepared sqlalchemy object into the db
        """    
        # Not doing batch inserts until we get single case to work properly.
        try:
            self.session.add(insertable['cluster'])
            self.session.commit()
        except IntegrityError:
            print('Prevented from inserting duplicate to cluster table')
            self.session.rollback()

        if insertable.get('gather'):
            try:
                self.session.add(insertable.get('gather'))
                self.session.commit()
            except IntegrityError:
                print('Prevented from inserting duplicate to gather table')
                self.session.rollback()

    def __call__(self, jsonin, path):
        path = path.rstrip('/')
        lnn = int(path.split('/')[-1].split('-')[-1])
        out = self.build_insertable(jsonin, self.find_entry(path), lnn)
        return out

SQL

2016-03-08 01:38:52,438 INFO sqlalchemy.engine.base.Engine SELECT gather.gather_id AS gather_gather_id, gather.cluster_guid AS gather_cluster_guid, gather.path AS gather_path, gather
.cluster_name AS gather_cluster_name, gather.gather_date AS gather_gather_date, gather.unfurl_start AS gather_unfurl_start, gather.unfurl_end AS gather_unfurl_end, gather.upload_date
AS gather_upload_date, gather.source_lnn AS gather_source_lnn, gather.last_full AS gather_last_full, gather.path_exists AS gather_path_exists, gather.type AS gather_type 
FROM gather 
WHERE gather.path = %(path_1)s 
LIMIT %(param_1)s
2016-03-08 01:38:52,438 INFO sqlalchemy.engine.base.Engine {'param_1': 1, 'path_1': '/mnt/logs/REALPAGE/2015-12-14-005'}
2016-03-08 01:38:52,440 INFO sqlalchemy.engine.base.Engine COMMIT
2016-03-08 01:38:52,441 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2016-03-08 01:38:52,441 INFO sqlalchemy.engine.base.Engine SELECT gather.gather_id AS gather_gather_id, gather.cluster_guid AS gather_cluster_guid, gather.path AS gather_path, gather
.cluster_name AS gather_cluster_name, gather.gather_date AS gather_gather_date, gather.unfurl_start AS gather_unfurl_start, gather.unfurl_end AS gather_unfurl_end, gather.upload_date
AS gather_upload_date, gather.source_lnn AS gather_source_lnn, gather.last_full AS gather_last_full, gather.path_exists AS gather_path_exists, gather.type AS gather_type 
FROM gather 
WHERE gather.gather_id = %(param_1)s
2016-03-08 01:38:52,441 INFO sqlalchemy.engine.base.Engine {'param_1': 'd284c3f7983f94bac95e024038820f05475feddb2f24aec2cb52d42c343194dd'}
2016-03-08 01:38:52,444 INFO sqlalchemy.engine.base.Engine INSERT INTO cluster (guid, site_id) VALUES (%(guid)s, %(site_id)s)
2016-03-08 01:38:52,444 INFO sqlalchemy.engine.base.Engine {'guid': '00074309e06ace7817523b06c7cbf76f7c08', 'site_id': None}
2016-03-08 01:38:52,445 INFO sqlalchemy.engine.base.Engine ROLLBACK

Notice that there isn't a commit between the select and the rollback.

Halp.