def select_chunk_into_outfile()

in core/lib/payload/copy.py [0:0]

• 29 lines of code
• 3 McCabe index (conditional complexity)

    def select_chunk_into_outfile(self, outfile, use_where):
        try:
            sql_string = sql.select_full_table_into_file_by_chunk(
                self.table_name,
                self.range_start_vars_array,
                self.range_end_vars_array,
                self._pk_for_filter,
                self.old_non_pk_column_list,
                self.select_chunk_size,
                use_where,
                self.where,
                self._idx_name_for_filter,
            )
            affected_rows = self.execute_sql(sql_string, (outfile,))
        except MySQLdb.OperationalError as e:
            errnum, errmsg = e.args
            # 1086: File exists
            if errnum == 1086:
                raise OSCError("FILE_ALREADY_EXIST", {"file": outfile})
            else:
                raise
        log.debug("{} affected".format(affected_rows))
        self.stats["outfile_lines"] = affected_rows + self.stats.setdefault(
            "outfile_lines", 0
        )
        self.stats["outfile_cnt"] = 1 + self.stats.setdefault("outfile_cnt", 0)
        self._cleanup_payload.add_file_entry(
            "{}.{}".format(self.outfile, self.outfile_suffix_end)
        )
        return affected_rows