db.py 文件源码

def __create_generic_notification(self, table=None, **notification):
        with self.acquire_conn() as conn:
            tx = conn.begin()
            try:
                result = conn.execute(table.insert(), **notification)

                tx.commit()
                logger.debug(
                    'notification_stored',
                    nid=result.inserted_primary_key,
                    notify_type=notification.get('notify_type)'))
                return True
            except (IntegrityError, SQLiteIntegrityError) as e:
                if is_duplicate_entry_error(e):
                    logger.debug(
                        '__create_generic_notification ignoring duplicate entry error')
                    return True
                else:
                    logger.exception('__create_generic_notification failed',
                                     **notification)
                    tx.rollback()
                    return False
            except BaseException:
                tx.rollback()
                logger.exception('__create_generic_notification failed', **notification)

            return False