def cached_revision_path()

in api_inference_community/hub.py [0:0]

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

def cached_revision_path(cache_dir, repo_id, revision) -> pathlib.Path:

    error_msg = f"No revision path found for {repo_id}, revision {revision}"

    if revision is None:
        revision = "main"

    repo_cache = _cached_repo_root_path(cache_dir, repo_id)

    if not repo_cache.is_dir():
        msg = f"Local repo {repo_cache} does not exist"
        logger.error(msg)
        raise Exception(msg)

    refs_dir = repo_cache / "refs"
    snapshots_dir = repo_cache / "snapshots"

    # Resolve refs (for instance to convert main to the associated commit sha)
    if refs_dir.is_dir():
        revision_file = refs_dir / revision
        if revision_file.exists():
            with revision_file.open() as f:
                revision = f.read()

    # Check if revision folder exists
    if not snapshots_dir.exists():
        msg = f"No local revision path {snapshots_dir} found for {repo_id}, revision {revision}"
        logger.error(msg)
        raise Exception(msg)

    cached_shas = os.listdir(snapshots_dir)
    if revision not in cached_shas:
        # No cache for this revision and we won't try to return a random revision
        logger.error(error_msg)
        raise Exception(error_msg)

    return snapshots_dir / revision