def list_checkpoint_dirs()

in src/hyperpod_nemo_adapter/utils/callbacks/local_ckpt_io.py [0:0]

• 13 lines of code
• 5 McCabe index (conditional complexity)

    def list_checkpoint_dirs(path):
        path = Path(path)
        path = format_s3_path(path) if is_s3_uri(path) else path
        fs = path_to_file_system(path)
        dirs = SageMakerLocalCheckpointIO.listdir(fs, path)
        candidate = []
        for f in dirs:
            step = SageMakerLocalCheckpointIO.load_step(fs, f)
            if step >= 0:
                candidate.append((step, f))

        candidate.sort(reverse=True)
        for s, f in candidate:
            yield s, fs.concat_path(f, _subdir())