def _writer_loop()

in ssiog/metrics_logger.py [0:0]

• 26 lines of code
• 10 McCabe index (conditional complexity)

    def _writer_loop(self):
        with open(self.file_name, "w", newline="") as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(["timestamp", "sample_lat"])

            while True:
                try:
                    metrics = []
                    while True:
                        metrics.append(self.queue.get_nowait())
                except queue.Empty:
                    pass

                if metrics:
                    writer.writerows(metrics)
                    csvfile.flush()
                    
                # Get all remaining metrics from the queue if shutdown.
                if self._shutdown:
                    try:
                        metrics = []
                        while True:
                            metrics.append(self.queue.get_nowait())
                    except queue.Empty:
                        pass
                    
                    if metrics:
                        writer.writerows(metrics)
                        csvfile.flush()
                    break

                time.sleep(self.flush_interval)