def has_write_access_to_repo()

in optimum/neuron/utils/cache_utils.py [0:0]

• 22 lines of code
• 9 McCabe index (conditional complexity)

def has_write_access_to_repo(repo_id: str) -> bool:
    # If the result has already been cached, use it instead of requesting the HF Hub again.
    token = get_token()
    key = (token, repo_id)
    if key in _CACHED_HAS_WRITE_ACCESS_TO_REPO:
        return _CACHED_HAS_WRITE_ACCESS_TO_REPO[key]

    api = HfApi()
    has_access = None
    try:
        api.delete_branch(repo_id=repo_id, repo_type="model", branch=f"this-branch-does-not-exist-{uuid4()}")
    except GatedRepoError:
        has_access = False
    except RepositoryNotFoundError:
        # We could raise an error to indicate the user that the repository could not even be found:
        # raise ValueError(f"Repository {repo_id} not found (repo_type: {repo_type}). Is it a private one?") from e
        # But here we simply return `False`, because it means that we do not have write access to this repo in the end.
        has_access = False
    except RevisionNotFoundError:
        has_access = True  # has write access, otherwise would have been 403 forbidden.
    except HfHubHTTPError as e:
        if e.response.status_code in (401, 403):
            has_access = False

    if has_access is None:
        raise ValueError(f"Cannot determine write access to {repo_id}")

    # Cache the result for subsequent calls.
    _CACHED_HAS_WRITE_ACCESS_TO_REPO[key] = has_access

    return has_access