Get data from the local db

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")

        # Skip Id field on insert
        if action == "insert" and "Id" in fields:
            del fields["Id"]

        # Build the list of fields to import
        columns = []
        columns.extend(fields.keys())
        columns.extend(lookups.keys())
        columns.extend(static.keys())

        if record_type:
            columns.append("RecordTypeId")
            # default to the profile assigned recordtype if we can't find any
            # query for the RT by developer name
            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

            # Add static values to row
            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)

            # Yield and start a new file every [batch_size] rows
            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

        # Yield result file for final batch
        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"]
            )
        )