(self, rows, use_shard_col=True)
Saves rows to a data source after deleting the old rows
        :param use_shard_col: use shard column along with doc id for searching rows to delete/update
Saves rows to a data source after deleting the old rows
        :param use_shard_col: use shard column along with doc id for searching rows to delete/update
Saves rows to a data source after deleting the old rows : param use_shard_col : use shard column along with doc id for searching rows to delete / update
def save_rows(self, rows, use_shard_col=True):
        """
        Saves rows to a data source after deleting the old rows
        :param use_shard_col: use shard column along with doc id for searching rows to delete/update
        """
        if not rows:
            return

        # transform format from ColumnValue to dict
        formatted_rows = [
            {i.column.database_column_name.decode('utf-8'): i.value for i in row}
            for row in rows
        ]
        doc_ids = set(row['doc_id'] for row in formatted_rows)
        table = self.get_table()
        if self.supports_upsert() and use_shard_col:
            queries = [self._upsert_query(table, formatted_rows)]
        else:
            delete = table.delete().where(table.c.doc_id.in_(doc_ids))
            # Using session.bulk_insert_mappings below might seem more inline
            #   with sqlalchemy API, but it results in
            #   appending an empty row which results in a postgres
            #   not-null constraint error, which has been hard to debug.
            # In addition, bulk_insert_mappings is less performant than
            #   the plain INSERT INTO VALUES statement resulting from below line
            #   because bulk_insert_mappings is meant for multi-table insertion
            #   so it has overhead of format conversions and multiple statements
            insert = table.insert().values(formatted_rows)
            queries = [delete, insert]
        with self.session_context() as session:
            for query in queries:
                session.execute(query)
def save_rows ( self , rows , use_shard_col = True ) : if not rows : return # transform format from ColumnValue to dict formatted_rows = [ { i . column . database_column_name . decode ( 'utf-8' ) : i . value for i in row } for row in rows ] doc_ids = set ( row [ 'doc_id' ] for row in formatted_rows ) table = self . get_table ( ) if self . supports_upsert ( ) and use_shard_col : queries = [ self . _upsert_query ( table , formatted_rows ) ] else : delete = table . delete ( ) . where ( table . c . doc_id . in_ ( doc_ids ) ) # Using session.bulk_insert_mappings below might seem more inline #   with sqlalchemy API, but it results in #   appending an empty row which results in a postgres #   not-null constraint error, which has been hard to debug. # In addition, bulk_insert_mappings is less performant than #   the plain INSERT INTO VALUES statement resulting from below line #   because bulk_insert_mappings is meant for multi-table insertion #   so it has overhead of format conversions and multiple statements insert = table . insert ( ) . values ( formatted_rows ) queries = [ delete , insert ] with self . session_context ( ) as session : for query in queries : session . execute ( query )