def process()

in basic_pitch/data/pipeline.py [0:0]

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

    def process(self, element: Any, *args: Tuple[Any, Any], **kwargs: Dict[str, Any]) -> None:
        if not isinstance(element, list):
            element = [element]

        logging.info(f"Writing to file batch of length {len(element)}")
        # hopefully uuids are unique enough
        with tf.io.TFRecordWriter(os.path.join(self.destination, f"{uuid.uuid4()}.tfrecord")) as writer:
            for example in element:
                writer.write(example.SerializeToString())