def download_from_oras(
    model: str,
    config_file: str,
    credentials_dir: str,
    save_dir: str,
    tokenizer_only: bool = False,
    concurrency: int = 10,
):
    """Download from an OCI-compliant registry using oras.

    Args:
        model (str): The name of the model to download.
        config_file (str): Path to a YAML config file that maps model and tokenizer names to registry paths.
        credentials_dir (str): Path to a directory containing credentials for the registry. It is expected to contain three
            files: `username`, `password`, and `registry`, each of which contains the corresponding credential.
        save_dir (str): Path to the directory where files will be downloaded.
        tokenizer_only (bool): If true, only download the tokenizer files.
        concurrency (int): The number of concurrent downloads to run.
    """
    if shutil.which(ORAS_CLI) is None:
        raise Exception(
            f'oras cli command `{ORAS_CLI}` is not found. Please install oras: https://oras.land/docs/installation ',
        )

    def _read_secrets_file(
        secret_file_path: str,
    ):
        try:
            with open(secret_file_path, encoding='utf-8') as f:
                return f.read().strip()
        except Exception as error:
            raise ValueError(
                f'secrets file {secret_file_path} failed to be read',
            ) from error

    secrets = {}
    for secret in ['username', 'password', 'registry']:
        secrets[secret] = _read_secrets_file(
            os.path.join(credentials_dir, secret),
        )

    with open(config_file, 'r', encoding='utf-8') as f:
        configs = yaml.safe_load(f.read())

    config_type = 'tokenizers' if tokenizer_only else 'models'
    path = configs[config_type][model]
    registry = secrets['registry']

    def get_oras_cmd(
        username: Optional[str] = None,
        password: Optional[str] = None,
    ):
        cmd = [
            ORAS_CLI,
            'pull',
            f'{registry}/{path}',
            '-o',
            save_dir,
            '--verbose',
            '--concurrency',
            str(concurrency),
        ]
        if username is not None:
            cmd.extend(['--username', username])
        if password is not None:
            cmd.extend(['--password', password])

        return cmd

    cmd_without_creds = get_oras_cmd()
    oras_cli_command = ' '.join(cmd_without_creds)
    log.info(f'CMD for oras cli to run: {oras_cli_command}')
    cmd_to_run = get_oras_cmd(
        username=secrets['username'],
        password=secrets['password'],
    )
    try:
        subprocess.run(cmd_to_run, check=True)
    except subprocess.CalledProcessError as e:
        # Intercept the error and replace the cmd, which may have sensitive info.
        raise subprocess.CalledProcessError(
            e.returncode,
            cmd_without_creds,
            e.output,
            e.stderr,
        )
Download from an OCI-compliant registry using oras.

    Args:
        model (str): The name of the model to download.
        config_file (str): Path to a YAML config file that maps model and tokenizer names to registry paths.
        credentials_dir (str): Path to a directory containing credentials for the registry. It is expected to contain three
            files: `username`, `password`, and `registry`, each of which contains the corresponding credential.
        save_dir (str): Path to the directory where files will be downloaded.
        tokenizer_only (bool): If true, only download the tokenizer files.
        concurrency (int): The number of concurrent downloads to run.
    
llmfoundry/utils/model_download_utils.py