def _get_batches(self, mapping, batch_size=10000):
    'Get data from the local db'
    action = mapping.get('action', 'insert')
    fields = mapping.get('fields', {}).copy()
    static = mapping.get('static', {})
    lookups = mapping.get('lookups', {})
    record_type = mapping.get('record_type')
    if ((action == 'insert') and ('Id' in fields)):
        del fields['Id']
    columns = []
    columns.extend(fields.keys())
    columns.extend(lookups.keys())
    columns.extend(static.keys())
    if record_type:
        columns.append('RecordTypeId')
        query = "SELECT Id FROM RecordType WHERE SObjectType='{0}'AND DeveloperName = '{1}' LIMIT 1"
        record_type_id = self.sf.query(query.format(mapping.get('sf_object'), record_type))['records'][0]['Id']
    query = self._query_db(mapping)
    total_rows = 0
    batch_num = 1

    def start_batch():
        batch_file = io.BytesIO()
        writer = unicodecsv.writer(batch_file)
        writer.writerow(columns)
        batch_ids = []
        return (batch_file, writer, batch_ids)
    (batch_file, writer, batch_ids) = start_batch()
    for row in query.yield_per(batch_size):
        total_rows += 1
        pkey = row[0]
        row = (list(row[1:]) + list(static.values()))
        if record_type:
            row.append(record_type_id)
        writer.writerow([self._convert(value) for value in row])
        batch_ids.append(pkey)
        if (not (total_rows % batch_size)):
            batch_file.seek(0)
            self.logger.info('    Processing batch {}'.format(batch_num))
            (yield (batch_file, batch_ids))
            (batch_file, writer, batch_ids) = start_batch()
            batch_num += 1
    if batch_ids:
        batch_file.seek(0)
        (yield (batch_file, batch_ids))
    self.logger.info('  Prepared {} rows for import to {}'.format(total_rows, mapping['sf_object']))
Get data from the local db
def _get_batches(self, mapping, batch_size=10000):
    
    action = mapping.get('action', 'insert')
    fields = mapping.get('fields', {}).copy()
    static = mapping.get('static', {})
    lookups = mapping.get('lookups', {})
    record_type = mapping.get('record_type')
    if ((action == 'insert') and ('Id' in fields)):
        del fields['Id']
    columns = []
    columns.extend(fields.keys())
    columns.extend(lookups.keys())
    columns.extend(static.keys())
    if record_type:
        columns.append('RecordTypeId')
        query = "SELECT Id FROM RecordType WHERE SObjectType='{0}'AND DeveloperName = '{1}' LIMIT 1"
        record_type_id = self.sf.query(query.format(mapping.get('sf_object'), record_type))['records'][0]['Id']
    query = self._query_db(mapping)
    total_rows = 0
    batch_num = 1

    def start_batch():
        batch_file = io.BytesIO()
        writer = unicodecsv.writer(batch_file)
        writer.writerow(columns)
        batch_ids = []
        return (batch_file, writer, batch_ids)
    (batch_file, writer, batch_ids) = start_batch()
    for row in query.yield_per(batch_size):
        total_rows += 1
        pkey = row[0]
        row = (list(row[1:]) + list(static.values()))
        if record_type:
            row.append(record_type_id)
        writer.writerow([self._convert(value) for value in row])
        batch_ids.append(pkey)
        if (not (total_rows % batch_size)):
            batch_file.seek(0)
            self.logger.info('    Processing batch {}'.format(batch_num))
            (yield (batch_file, batch_ids))
            (batch_file, writer, batch_ids) = start_batch()
            batch_num += 1
    if batch_ids:
        batch_file.seek(0)
        (yield (batch_file, batch_ids))
    self.logger.info('  Prepared {} rows for import to {}'.format(total_rows, mapping['sf_object']))