File size: 5,907 Bytes
76d6ddf
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# Copyright 2026 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""CLI error handling utilities."""

import traceback
from collections.abc import Callable

from huggingface_hub.errors import (
    BucketNotFoundError,
    CLIError,
    CLIExtensionInstallError,
    DeviceCodeError,
    EntryNotFoundError,
    GatedRepoError,
    HfHubHTTPError,
    HfUriError,
    IncompleteSnapshotError,
    LocalEntryNotFoundError,
    LocalTokenNotFoundError,
    OfflineModeIsEnabled,
    OIDCError,
    RemoteEntryNotFoundError,
    RepositoryNotFoundError,
    RevisionNotFoundError,
)


def _format_repo_not_found(error: RepositoryNotFoundError) -> str:
    label = error.repo_type.capitalize() if error.repo_type else "Repository"
    if error.repo_id:
        msg = f"{label} '{error.repo_id}' not found."
    else:
        msg = f"{label} not found."
    msg += "\nIf the repo is private, make sure you are authenticated and your token has the required permissions."

    msg += "\nIf the repo does not exist, create it with: "
    if error.repo_id is not None:
        type_flag = f" --type {error.repo_type}" if error.repo_type and error.repo_type != "model" else ""
        msg += f"hf repos create {error.repo_id}{type_flag}"
    else:
        msg += "hf repos create <repo_id>"

    return msg


def _format_gated_repo(error: GatedRepoError) -> str:
    label = error.repo_type if error.repo_type else "repository"
    if error.repo_id:
        return f"Access denied. {label.capitalize()} '{error.repo_id}' requires approval."
    return f"Access denied. This {label} requires approval."


def _format_bucket_not_found(error: BucketNotFoundError) -> str:
    if error.bucket_id:
        msg = f"Bucket '{error.bucket_id}' not found."
        cmd = f"hf buckets create {error.bucket_id}"
    else:
        msg = "Bucket not found."
        cmd = "hf buckets create <bucket_id>"
    msg += "\nIf the bucket is private, make sure you are authenticated and your token has the required permissions."
    msg += f"\nIf the bucket does not exist, create it with: {cmd}"
    return msg


def _format_entry_not_found(error: RemoteEntryNotFoundError) -> str:
    label = error.repo_type if error.repo_type else "repository"
    url = str(error.response.url) if error.response else None
    if error.repo_id:
        msg = f"File not found in {label} '{error.repo_id}'."
    else:
        msg = f"File not found in {label}."
    if url:
        msg += f"\nURL: {url}"
    return msg


def _format_local_entry_not_found(error: LocalEntryNotFoundError) -> str:
    cause = error.__cause__
    if cause is not None:
        return f"Local entry not found. {cause}"
    return f"Local entry not found. {error}"


def _format_incomplete_snapshot(error: IncompleteSnapshotError) -> str:
    msg = _format_local_entry_not_found(error)
    msg += f"\nIncomplete snapshot available at: {error.snapshot_path}"
    return msg


def _format_revision_not_found(error: RevisionNotFoundError) -> str:
    label = error.repo_type if error.repo_type else "repository"
    if error.repo_id:
        return f"Revision not found in {label} '{error.repo_id}'."
    return f"Revision not found in {label}. Check the revision parameter."


def _format_cli_error(error: CLIError) -> str:
    """No traceback, just the error message."""
    return str(error)


def _format_cli_extension_install_error(error: CLIExtensionInstallError) -> str:
    """Format a CLI extension installation error.

    The error is likely to be a tricky subprocess error to investigate. In this specific case we want to format the
    traceback of the root cause while keeping the "nicely formatted" error message of the CLIExtensionInstallError
    as a 1-line message.
    """
    cause_tb = (
        "".join(traceback.format_exception(type(error.__cause__), error.__cause__, error.__cause__.__traceback__))
        if error.__cause__ is not None
        else ""
    )
    return f"{cause_tb}\n{error}"


CLI_ERROR_MAPPINGS: dict[type[Exception], Callable[..., str]] = {
    OfflineModeIsEnabled: lambda error: str(error),
    # GatedRepoError must come before RepositoryNotFoundError (it's a subclass).
    GatedRepoError: _format_gated_repo,
    BucketNotFoundError: _format_bucket_not_found,
    RepositoryNotFoundError: _format_repo_not_found,
    RevisionNotFoundError: _format_revision_not_found,
    LocalTokenNotFoundError: lambda _: "Not logged in. Run 'hf auth login' first.",
    OIDCError: lambda error: f"OIDC Exchange failed. {error}",
    DeviceCodeError: lambda error: f"Login failed: {error}",
    RemoteEntryNotFoundError: _format_entry_not_found,
    # IncompleteSnapshotError must come before LocalEntryNotFoundError (it's a subclass).
    IncompleteSnapshotError: _format_incomplete_snapshot,
    LocalEntryNotFoundError: _format_local_entry_not_found,
    EntryNotFoundError: lambda error: str(error),
    HfHubHTTPError: lambda error: str(error),
    HfUriError: lambda error: f"Invalid HF URI: {error.uri}. {error.msg}",
    ValueError: lambda error: f"Invalid value. {error}",
    CLIExtensionInstallError: _format_cli_extension_install_error,
    CLIError: _format_cli_error,
}


def format_known_exception(error: Exception) -> str | None:
    for exc_type, formatter in CLI_ERROR_MAPPINGS.items():
        if isinstance(error, exc_type):
            return formatter(error)
    return None