src/dirbs/listgen/generator.py
def _write_csv_lists(self):  # noqa: C901
        """Write full CSV lists from the intermediate tables."""
        with futures.ProcessPoolExecutor(max_workers=self._nworkers) as executor:
            # We use ProcessPoolExecutor as these tasks are CPU intensive. This means that we do not have access
            # to the self._conn, self._metadata_conn and self._logger objects within the functions (they are removed
            # during pickling, see __getstate__)
            #
            futures_to_cb = {}
            md = defaultdict(list)
            self._queue_csv_writer_job(executor, futures_to_cb, self._write_delta_csv_blacklist, 'delta blacklist', md)
            for op in self._operators:
                for fn, desc in \
                        [(self._write_delta_csv_notifications_list, 'delta notifications list for operator {0}'),
                         (self._write_delta_csv_exceptions_list, 'delta exceptions list for operator {0}')]:
                    self._queue_csv_writer_job(executor, futures_to_cb, partial(fn, op.id), desc.format(op.id), md)

            if not self._no_full_lists:
                for fn, desc in [(self._write_full_csv_blacklist, 'full blacklist')]:
                    self._queue_csv_writer_job(executor, futures_to_cb, fn, desc, md)
                for op in self._operators:
                    for fn, desc in \
                            [(self._write_full_csv_notifications_list, 'full notifications list for operator {0}'),
                             (self._write_full_csv_exceptions_list, 'full exceptions list for operator {0}')]:
                        self._queue_csv_writer_job(executor, futures_to_cb, partial(fn, op.id), desc.format(op.id), md)

            self._wait_for_futures(futures_to_cb)

        metadata.add_optional_job_metadata(self._metadata_conn, 'dirbs-listgen', self._run_id, **md)

        self._logger.info('Zipping up lists...')
        with zipfile.ZipFile(os.path.join(self._output_dir, '{0}_blacklist.zip'.format(self._date_str)), 'w') as zf:
            for csv_path in glob.glob(os.path.join(self._output_dir, '{0}_blacklist*.csv'.format(self._date_str))):
                zf.write(csv_path, arcname=os.path.basename(csv_path))
                os.remove(csv_path)

        for op in self._operators:
            with zipfile.ZipFile(os.path.join(self._output_dir,
                                              '{0}_notifications_{1}.zip'.format(self._date_str, op.id)), 'w') as zf:
                for csv_path in glob.glob(os.path.join(self._output_dir,
                                                       '{0}_notifications_{1}*.csv'.format(self._date_str, op.id))):
                    zf.write(csv_path, arcname=os.path.basename(csv_path))
                    os.remove(csv_path)
            with zipfile.ZipFile(os.path.join(self._output_dir,
                                              '{0}_exceptions_{1}.zip'.format(self._date_str, op.id)), 'w') as zf:
                for csv_path in glob.glob(os.path.join(self._output_dir,
                                                       '{0}_exceptions_{1}*.csv'.format(self._date_str, op.id))):
                    zf.write(csv_path, arcname=os.path.basename(csv_path))
                    os.remove(csv_path)
        self._logger.info('Zipped up lists')
Write full CSV lists from the intermediate tables.
Write full CSV lists from the intermediate tables.
Write full CSV lists from the intermediate tables .
def _write_csv_lists(self):  
        with futures.ProcessPoolExecutor(max_workers=self._nworkers) as executor:
            futures_to_cb = {}
            md = defaultdict(list)
            self._queue_csv_writer_job(executor, futures_to_cb, self._write_delta_csv_blacklist, 'delta blacklist', md)
            for op in self._operators:
                for fn, desc in \
                        [(self._write_delta_csv_notifications_list, 'delta notifications list for operator {0}'),
                         (self._write_delta_csv_exceptions_list, 'delta exceptions list for operator {0}')]:
                    self._queue_csv_writer_job(executor, futures_to_cb, partial(fn, op.id), desc.format(op.id), md)
            if not self._no_full_lists:
                for fn, desc in [(self._write_full_csv_blacklist, 'full blacklist')]:
                    self._queue_csv_writer_job(executor, futures_to_cb, fn, desc, md)
                for op in self._operators:
                    for fn, desc in \
                            [(self._write_full_csv_notifications_list, 'full notifications list for operator {0}'),
                             (self._write_full_csv_exceptions_list, 'full exceptions list for operator {0}')]:
                        self._queue_csv_writer_job(executor, futures_to_cb, partial(fn, op.id), desc.format(op.id), md)
            self._wait_for_futures(futures_to_cb)
        metadata.add_optional_job_metadata(self._metadata_conn, 'dirbs-listgen', self._run_id, **md)
        self._logger.info('Zipping up lists...')
        with zipfile.ZipFile(os.path.join(self._output_dir, '{0}_blacklist.zip'.format(self._date_str)), 'w') as zf:
            for csv_path in glob.glob(os.path.join(self._output_dir, '{0}_blacklist*.csv'.format(self._date_str))):
                zf.write(csv_path, arcname=os.path.basename(csv_path))
                os.remove(csv_path)
        for op in self._operators:
            with zipfile.ZipFile(os.path.join(self._output_dir,
                                              '{0}_notifications_{1}.zip'.format(self._date_str, op.id)), 'w') as zf:
                for csv_path in glob.glob(os.path.join(self._output_dir,
                                                       '{0}_notifications_{1}*.csv'.format(self._date_str, op.id))):
                    zf.write(csv_path, arcname=os.path.basename(csv_path))
                    os.remove(csv_path)
            with zipfile.ZipFile(os.path.join(self._output_dir,
                                              '{0}_exceptions_{1}.zip'.format(self._date_str, op.id)), 'w') as zf:
                for csv_path in glob.glob(os.path.join(self._output_dir,
                                                       '{0}_exceptions_{1}*.csv'.format(self._date_str, op.id))):
                    zf.write(csv_path, arcname=os.path.basename(csv_path))
                    os.remove(csv_path)
        self._logger.info('Zipped up lists')
def _write_csv_lists ( self ) : with futures . ProcessPoolExecutor ( max_workers = self . _nworkers ) as executor : futures_to_cb = { } md = defaultdict ( list ) self . _queue_csv_writer_job ( executor , futures_to_cb , self . _write_delta_csv_blacklist , 'delta blacklist' , md ) for op in self . _operators : for fn , desc in [ ( self . _write_delta_csv_notifications_list , 'delta notifications list for operator {0}' ) , ( self . _write_delta_csv_exceptions_list , 'delta exceptions list for operator {0}' ) ] : self . _queue_csv_writer_job ( executor , futures_to_cb , partial ( fn , op . id ) , desc . format ( op . id ) , md ) if not self . _no_full_lists : for fn , desc in [ ( self . _write_full_csv_blacklist , 'full blacklist' ) ] : self . _queue_csv_writer_job ( executor , futures_to_cb , fn , desc , md ) for op in self . _operators : for fn , desc in [ ( self . _write_full_csv_notifications_list , 'full notifications list for operator {0}' ) , ( self . _write_full_csv_exceptions_list , 'full exceptions list for operator {0}' ) ] : self . _queue_csv_writer_job ( executor , futures_to_cb , partial ( fn , op . id ) , desc . format ( op . id ) , md ) self . _wait_for_futures ( futures_to_cb ) metadata . add_optional_job_metadata ( self . _metadata_conn , 'dirbs-listgen' , self . _run_id , ** md ) self . _logger . info ( 'Zipping up lists...' ) with zipfile . ZipFile ( os . path . join ( self . _output_dir , '{0}_blacklist.zip' . format ( self . _date_str ) ) , 'w' ) as zf : for csv_path in glob . glob ( os . path . join ( self . _output_dir , '{0}_blacklist*.csv' . format ( self . _date_str ) ) ) : zf . write ( csv_path , arcname = os . path . basename ( csv_path ) ) os . remove ( csv_path ) for op in self . _operators : with zipfile . ZipFile ( os . path . join ( self . _output_dir , '{0}_notifications_{1}.zip' . format ( self . _date_str , op . id ) ) , 'w' ) as zf : for csv_path in glob . glob ( os . path . join ( self . _output_dir , '{0}_notifications_{1}*.csv' . format ( self . _date_str , op . id ) ) ) : zf . write ( csv_path , arcname = os . path . basename ( csv_path ) ) os . remove ( csv_path ) with zipfile . ZipFile ( os . path . join ( self . _output_dir , '{0}_exceptions_{1}.zip' . format ( self . _date_str , op . id ) ) , 'w' ) as zf : for csv_path in glob . glob ( os . path . join ( self . _output_dir , '{0}_exceptions_{1}*.csv' . format ( self . _date_str , op . id ) ) ) : zf . write ( csv_path , arcname = os . path . basename ( csv_path ) ) os . remove ( csv_path ) self . _logger . info ( 'Zipped up lists' )
Write full CSV lists from the intermediate tables.
Write full CSV lists from the intermediate tables .
# noqa: C901 """Write full CSV lists from the intermediate tables.""" # We use ProcessPoolExecutor as these tasks are CPU intensive. This means that we do not have access # to the self._conn, self._metadata_conn and self._logger objects within the functions (they are removed # during pickling, see __getstate__) #
self