def _perform_io_writes()

in s3transfer/__init__.py [0:0]

• 23 lines of code
• 4 McCabe index (conditional complexity)

    def _perform_io_writes(self, filename):
        with self._os.open(filename, 'wb') as f:
            while True:
                task = self._ioqueue.get()
                if task is SHUTDOWN_SENTINEL:
                    logger.debug(
                        "Shutdown sentinel received in IO handler, "
                        "shutting down IO handler."
                    )
                    return
                else:
                    try:
                        offset, data = task
                        f.seek(offset)
                        f.write(data)
                    except Exception as e:
                        logger.debug(
                            "Caught exception in IO thread: %s",
                            e,
                            exc_info=True,
                        )
                        self._ioqueue.trigger_shutdown()
                        raise