codekingpro/portable-devtools
114k
1# Copyright 2022 The HuggingFace Team. All rights reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14"""Contains CLI utilities (styling, helpers)."""15 16import dataclasses17import datetime18import difflib19import importlib.metadata20import json21import os22import re23import subprocess24import sys25import time26from collections.abc import Callable, Sequence27from enum import Enum28from pathlib import Path29from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeVar, Union, cast30 31import click32import typer33from typer.core import TyperCommand, TyperGroup34 35from huggingface_hub import Volume, __version__, constants36from huggingface_hub.errors import CLIError37from huggingface_hub.utils import (38 get_session,39 hf_raise_for_status,40 installation_method,41 logging,42 tabulate,43)44from huggingface_hub.utils._dotenv import load_dotenv45 46from ._output import OutputFormatWithAuto, out47 48 49logger = logging.get_logger()50 51# Arbitrary maximum length of a cell in a table output52_MAX_CELL_LENGTH = 3553 54# Arbitrary default limit for models/datasets/spaces list commands.55REPO_LIST_DEFAULT_LIMIT = 3056 57if TYPE_CHECKING:58 from huggingface_hub.hf_api import HfApi59 60 61def get_hf_api(token: str | None = None) -> "HfApi":62 # Import here to avoid circular import63 from huggingface_hub.hf_api import HfApi64 65 return HfApi(token=token, library_name="huggingface-cli", library_version=__version__)66 67 68#### TYPER UTILS69 70CLI_REFERENCE_URL = "https://huggingface.co/docs/huggingface_hub/en/guides/cli"71 72 73def generate_epilog(examples: list[str], docs_anchor: str | None = None) -> str:74 """Generate an epilog with examples and a Learn More section.75 76 Args:77 examples: List of example commands (without the `$ ` prefix).78 docs_anchor: Optional anchor for the docs URL (e.g., "#hf-download").79 80 Returns:81 Formatted epilog string.82 """83 docs_url = f"{CLI_REFERENCE_URL}{docs_anchor}" if docs_anchor else CLI_REFERENCE_URL84 examples_str = "\n".join(f" $ {ex}" for ex in examples)85 return f"""\86Examples87{examples_str}88 89Learn more90 Use `hf <command> --help` for more information about a command.91 Read the documentation at {docs_url}92"""93 94 95TOPIC_T = Literal["main", "help"] | str96FallbackHandlerT = Callable[[list[str], set[str]], int | None]97ExpandPropertyT = TypeVar("ExpandPropertyT", bound=str)98 99 100def _format_epilog_no_indent(epilog: str | None, ctx: click.Context, formatter: click.HelpFormatter) -> None:101 """Write the epilog without indentation."""102 if epilog:103 formatter.write_paragraph()104 for line in epilog.split("\n"):105 formatter.write_text(line)106 107 108_ALIAS_SPLIT = re.compile(r"\s*\|\s*")109 110 111class HFCliTyperGroup(TyperGroup):112 """113 Typer Group that:114 - lists commands alphabetically within sections.115 - separates commands by topic (main, help, etc.).116 - formats epilog without extra indentation.117 - supports aliases via pipe-separated names (e.g. ``name="list | ls"``).118 - consumes the global formatting flags (``--format``, ``--json``, ``-q`` / ``--quiet``)119 anywhere in the args of a leaf command and applies them to ``out``, so leaf120 commands don't need to declare these options themselves.121 - rewrites ``spaces/user/repo`` to ``user/repo --type space`` for commands that accept ``--type``.122 - enriches "No such option" / "No such command" errors with available options or commands.123 """124 125 def invoke(self, ctx: click.Context) -> None:126 """Enrich unknown-option errors with available options or subcommands.127 128 Catches `NoSuchOption` raised during subcommand `make_context()`129 (option parsing). For leaf commands (e.g. `hf repos create --test`)130 we list the command's options; for groups (e.g. `hf cache --test`)131 we list subcommands since groups have no user-facing options.132 """133 try:134 return super().invoke(ctx)135 except click.NoSuchOption as e:136 if e.ctx is not None and e.ctx.command is not None:137 cmd = e.ctx.command138 if isinstance(cmd, click.Group):139 # Group has no user-facing options -> show subcommands instead140 items = [141 (name, sub.get_short_help_str(limit=80))142 for name in cmd.list_commands(e.ctx)143 if (sub := cmd.get_command(e.ctx, name)) is not None and not sub.hidden144 ]145 _enrich_usage_error(e, "commands", items)146 else:147 # Leaf command -> show its options using Click's rich formatting148 items = [149 record150 for p in cmd.get_params(e.ctx)151 if isinstance(p, click.Option) and not p.hidden and (record := p.get_help_record(e.ctx))152 ]153 _enrich_usage_error(e, "options", items)154 raise155 156 def resolve_command(self, ctx: click.Context, args: list[str]) -> tuple:157 cmd_name = args[0] if args and not args[0].startswith("-") else None158 cmd = self.get_command(ctx, cmd_name) if cmd_name else None159 160 if cmd is not None:161 self._rewrite_repo_type_prefix(cmd, args)162 163 try:164 name, resolved_cmd, sub_args = super().resolve_command(ctx, args)165 except click.UsageError as e:166 # Unknown subcommand -> add fuzzy suggestions and list available commands.167 if cmd is None and cmd_name is not None:168 # Expand aliases ("list | ls" → ["list", "ls"]) for accurate fuzzy matching.169 visible_names = [170 alias171 for key, registered in self.commands.items()172 if not registered.hidden173 for alias in _ALIAS_SPLIT.split(key)174 ]175 matches = difflib.get_close_matches(cmd_name, visible_names)176 if matches:177 suggestions = ", ".join(f"'{m}'" for m in matches)178 e.message = f"{e.message.rstrip('.')}. Did you mean {suggestions}?"179 items = [180 (name, sub.get_short_help_str(limit=80))181 for name in self.list_commands(ctx)182 if (sub := self.get_command(ctx, name)) is not None and not sub.hidden183 ]184 _enrich_usage_error(e, "commands", items)185 raise186 187 # If we just resolved a leaf command, eagerly consume any global formatting188 # flags (--format / --json / -q / --quiet) from its args before click parses189 # them. Group resolution is recursive — leaves (and only leaves) need this.190 if resolved_cmd is not None and not isinstance(resolved_cmd, click.Group):191 _consume_format_flags_for_leaf(resolved_cmd, sub_args)192 193 return name, resolved_cmd, sub_args194 195 @staticmethod196 def _rewrite_repo_type_prefix(cmd: click.Command, args: list[str]) -> None:197 """Rewrite prefixed repo IDs (e.g. ``spaces/user/repo``) to ``user/repo --type space``.198 199 Only applies to commands that have a ``--type`` / ``--repo-type`` option and200 at least one repo-ID positional argument (any ``click.Argument`` whose name201 ends with ``_id``, e.g. ``repo_id``, ``from_id``, ``to_id``). When the202 token that maps to such an argument matches ``{prefix}/org/repo`` (where203 *prefix* is one of ``spaces``, ``datasets``, or ``models``), the prefix is204 stripped and an implicit ``--type {type}`` is appended. An error is raised205 if ``--type`` is also provided explicitly or if multiple prefixed arguments206 disagree on the repo type.207 208 Only repo-ID positional slots are inspected so that other positional209 arguments (filenames, local paths, patterns …) are never misinterpreted as210 prefixed repo IDs.211 """212 has_type_option = any(isinstance(param, click.Option) and "--type" in param.opts for param in cmd.params)213 if not has_type_option:214 return215 216 # Locate all repo-ID positional arguments and their indices among Arguments.217 repo_id_positions: set[int] = set()218 arg_idx = 0219 for param in cmd.params:220 if isinstance(param, click.Argument):221 if param.name in ("repo_id", "from_id", "to_id"):222 repo_id_positions.add(arg_idx)223 arg_idx += 1224 225 if not repo_id_positions:226 return227 228 # Build a set of option names that consume a following value token.229 value_options: set[str] = set()230 for param in cmd.params:231 if isinstance(param, click.Option) and not param.is_flag:232 for opt in (*param.opts, *param.secondary_opts):233 value_options.add(opt)234 235 # Walk through args (skipping args[0] = command name) to map positional236 # slots to their indices in `args`.237 positional_count = 0238 repo_id_arg_indices: list[int] = []239 i = 1240 while i < len(args):241 arg = args[i]242 if arg == "--":243 break # everything after -- is positional literal; stop rewriting244 if arg.startswith("-"):245 if "=" in arg or arg not in value_options:246 i += 1 # flag or --opt=val — single token247 else:248 i += 2 # value-taking option — skip the value too249 else:250 if positional_count in repo_id_positions:251 repo_id_arg_indices.append(i)252 positional_count += 1253 i += 1254 255 if not repo_id_arg_indices:256 return257 258 # Check each repo-ID arg for a type prefix and collect rewrites.259 inferred_type: str | None = None260 first_prefix: str | None = None261 rewrites: list[tuple[int, str]] = [] # (args index, new value without prefix)262 263 for arg_index in repo_id_arg_indices:264 parts = args[arg_index].split("/", 2)265 if len(parts) != 3 or parts[0] not in constants.REPO_TYPES_MAPPING:266 continue267 prefix = parts[0]268 mapped_type = constants.REPO_TYPES_MAPPING[prefix]269 if inferred_type is not None and mapped_type != inferred_type:270 raise click.UsageError(f"Conflicting repo type prefixes: '{first_prefix}/' and '{prefix}/'.")271 inferred_type = mapped_type272 first_prefix = prefix273 rewrites.append((arg_index, f"{parts[1]}/{parts[2]}"))274 275 if not rewrites:276 return277 278 # Error if --type / --repo-type was also provided explicitly.279 if any(280 arg == "--type" or arg.startswith("--type=") or arg == "--repo-type" or arg.startswith("--repo-type=")281 for arg in args282 ):283 raise click.UsageError(284 f"Ambiguous repo type: got prefix '{first_prefix}/' in repo ID and explicit --type. Use one or the other."285 )286 287 # Apply all rewrites and append --type once.288 for arg_index, new_value in rewrites:289 args[arg_index] = new_value290 args.extend(["--type", inferred_type]) # type: ignore291 292 def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:293 # Try exact match first294 cmd = super().get_command(ctx, cmd_name)295 if cmd is not None:296 return cmd297 # Fall back to alias lookup: check if cmd_name matches any alias298 # taken from https://github.com/fastapi/typer/issues/132#issuecomment-2417492805299 for registered_name, registered_cmd in self.commands.items():300 aliases = _ALIAS_SPLIT.split(registered_name)301 if cmd_name in aliases:302 return registered_cmd303 return None304 305 def _alias_map(self) -> dict[str, list[str]]:306 """Build a mapping from primary command name to its aliases (if any)."""307 result: dict[str, list[str]] = {}308 for registered_name in self.commands:309 parts = _ALIAS_SPLIT.split(registered_name)310 primary = parts[0]311 result[primary] = parts[1:]312 return result313 314 def format_commands(self, ctx: click.Context, formatter: click.HelpFormatter) -> None:315 topics: dict[str, list] = {}316 alias_map = self._alias_map()317 318 for name in self.list_commands(ctx):319 cmd = self.get_command(ctx, name)320 if cmd is None or cmd.hidden:321 continue322 help_text = cmd.get_short_help_str(limit=formatter.width)323 aliases = alias_map.get(name, [])324 if aliases:325 help_text = f"{help_text} [alias: {', '.join(aliases)}]"326 topic = getattr(cmd, "topic", "main")327 topics.setdefault(topic, []).append((name, help_text))328 329 with formatter.section("Main commands"):330 formatter.write_dl(topics["main"])331 for topic in sorted(topics.keys()):332 if topic == "main":333 continue334 with formatter.section(f"{topic.capitalize()} commands"):335 formatter.write_dl(topics[topic])336 337 def format_epilog(self, ctx: click.Context, formatter: click.HelpFormatter) -> None:338 # Collect only the first example from each command (to keep group help concise)339 # Full examples are shown in individual subcommand help (e.g. `hf buckets sync --help`)340 all_examples: list[str] = []341 for name in self.list_commands(ctx):342 cmd = self.get_command(ctx, name)343 if cmd is None or cmd.hidden:344 continue345 cmd_examples = getattr(cmd, "examples", [])346 if cmd_examples:347 all_examples.append(cmd_examples[0])348 349 if all_examples:350 epilog = generate_epilog(all_examples)351 _format_epilog_no_indent(epilog, ctx, formatter)352 elif self.epilog:353 _format_epilog_no_indent(self.epilog, ctx, formatter)354 355 def list_commands(self, ctx: click.Context) -> list[str]: # type: ignore[name-defined]356 # For aliased commands ("list | ls"), use the primary name (first entry).357 primary_names: list[str] = []358 for name in self.commands:359 primary = _ALIAS_SPLIT.split(name)[0]360 primary_names.append(primary)361 return sorted(primary_names)362 363 364_FORMATTING_OPTIONS_HELP_RECORDS: list[tuple[str, str]] = [365 (366 "--format [auto|human|agent|json|quiet]",367 "Output format. Defaults to 'auto' which picks 'agent' or 'human' based on the terminal.",368 ),369 ("--json", "JSON output. Equivalent to '--format json'."),370 ("-q, --quiet", "Quiet output (one ID per line). Equivalent to '--format quiet'."),371]372 373 374def _format_formatting_options_section(formatter: click.HelpFormatter) -> None:375 with formatter.section("Formatting options"):376 formatter.write_dl(_FORMATTING_OPTIONS_HELP_RECORDS)377 378 379def _has_local_formatting_option(cmd: click.Command) -> bool:380 """Return True if the command defines its own --format, --json or --quiet / -q.381 382 Used to skip the global formatting flag pre-processor and the duplicated "Formatting options" help section for383 legacy commands like 'hf jobs ps' that have their own format/quiet options.384 """385 for param in cmd.params:386 if not isinstance(param, click.Option):387 continue388 opts = (*param.opts, *param.secondary_opts)389 if "--format" in opts or "--json" in opts or "--quiet" in opts or "-q" in opts:390 return True391 return False392 393 394def _consume_format_flags_for_leaf(cmd: click.Command, args: list[str]) -> None:395 """Apply global formatting flags from 'args' to a leaf command.396 397 Two modes, depending on the command:398 399 * **Pass-through commands** (ignore_unknown_options=True, e.g. 'hf extensions exec'):400 args are forwarded verbatim to an external binary; we don't touch them.401 402 * **Legacy commands with a local --format option** (e.g. 'hf jobs ps' whose '--format' accepts Go templates):403 the global flags are rewritten in-place to the legacy form ('--json' → '--format json', '--quiet'/'-q' → '--format quiet'404 when the cmd has no own '--quiet') so click can parse them locally. This preserves backwards compatibility with the previous shorthand behavior.405 406 * **Modern commands** (no local format/quiet/json options): the flags '--format <value>' / '--json' / '--quiet' / '-q' are stripped from 'args' and applied to the singleton 'out'.407 408 Raises click.UsageError if multiple conflicting flags are supplied (e.g. '--json' together with '--format table').409 """410 if cmd.context_settings.get("ignore_unknown_options"):411 return412 413 has_local_format = False414 has_local_quiet = False415 has_local_json = False416 for param in cmd.params:417 if not isinstance(param, click.Option):418 continue419 opts = (*param.opts, *param.secondary_opts)420 if "--format" in opts:421 has_local_format = True422 if "--quiet" in opts or "-q" in opts:423 has_local_quiet = True424 if "--json" in opts:425 has_local_json = True426 427 if has_local_format:428 _rewrite_legacy_shorthands(args, rewrite_json=not has_local_json, rewrite_quiet=not has_local_quiet)429 return430 431 # Strip --format/--json/-q/--quiet from 'args' and apply to 'out'432 chosen_mode: OutputFormatWithAuto = OutputFormatWithAuto.auto433 chosen_flag: str | None = None434 435 def _check_conflict(new_flag: str) -> None:436 # Reject any second formatting flag before parsing values, so the user gets437 # a "mutually exclusive" error rather than e.g. an "invalid value" error438 # from the second flag's argument.439 if chosen_flag is not None:440 raise click.UsageError(f"'{chosen_flag}' and '{new_flag}' are mutually exclusive.")441 442 i = 0443 while i < len(args):444 arg = args[i]445 if arg == "--":446 break # everything after '--' is a positional literal447 if arg == "--format":448 _check_conflict("--format")449 if i + 1 >= len(args):450 raise click.UsageError("Option '--format' requires a value.")451 chosen_mode = _parse_format_value(args[i + 1])452 chosen_flag = "--format"453 del args[i : i + 2] # --format value => 2 args removed454 continue455 if arg.startswith("--format="):456 _check_conflict("--format")457 chosen_mode = _parse_format_value(arg[len("--format=") :])458 chosen_flag = "--format"459 del args[i : i + 1]460 continue461 if arg == "--json":462 _check_conflict("--json")463 chosen_mode = OutputFormatWithAuto.json464 chosen_flag = "--json"465 del args[i : i + 1]466 continue467 if arg in ("-q", "--quiet"):468 _check_conflict(arg)469 chosen_mode = OutputFormatWithAuto.quiet470 chosen_flag = arg471 del args[i : i + 1]472 continue473 i += 1474 475 out.set_mode(chosen_mode)476 477 478def _rewrite_legacy_shorthands(args: list[str], *, rewrite_json: bool, rewrite_quiet: bool) -> None:479 """Rewrite --json / -q / --quiet to --format ... for legacy commands.480 481 Used for commands like 'hf jobs ps' that still own their '--format' option.482 The rewrite lets users keep using the global shorthand while click parses483 '--format <value>' locally.484 """485 has_format_in_args = any(arg == "--format" or arg.startswith("--format=") for arg in args)486 487 if rewrite_json and "--json" in args:488 if has_format_in_args:489 raise click.UsageError("'--json' and '--format' are mutually exclusive.")490 idx = args.index("--json")491 args[idx : idx + 1] = ["--format", "json"]492 has_format_in_args = True493 494 if rewrite_quiet:495 flag = "-q" if "-q" in args else ("--quiet" if "--quiet" in args else None)496 if flag is not None:497 if has_format_in_args:498 raise click.UsageError(f"'{flag}' and '--format' are mutually exclusive.")499 idx = args.index(flag)500 args[idx : idx + 1] = ["--format", "quiet"]501 502 503def _parse_format_value(value: str) -> "OutputFormatWithAuto":504 try:505 return OutputFormatWithAuto(value)506 except ValueError:507 valid = ", ".join(m.value for m in OutputFormatWithAuto)508 raise click.UsageError(f"Invalid value for '--format': '{value}'. Valid values: {valid}.") from None509 510 511def _enrich_usage_error(error: click.UsageError, label: str, items: list[tuple[str, str]]) -> None:512 """Append a list of available options or commands to a usage error message."""513 if not items or error.ctx is None or f"Available {label} for" in error.message:514 return515 cmd_path = error.ctx.command_path516 lines = [f"\n\nAvailable {label} for '{cmd_path}':"]517 for name, help_text in items:518 lines.append(f" {name:30s} {help_text}")519 lines.append(f"\nRun '{cmd_path} --help' for full details.")520 if isinstance(error, click.NoSuchOption) and error.possibilities:521 lines.append(f"\nDid you mean: {', '.join(sorted(error.possibilities))}?")522 error.possibilities = []523 error.message += "\n".join(lines)524 525 526def fallback_typer_group_factory(527 fallback_handler: FallbackHandlerT,528 extra_commands_provider: Callable[[], list[tuple[str, str]]] | None = None,529) -> type[HFCliTyperGroup]:530 """Return a Typer group class that runs a fallback handler before command resolution."""531 532 class FallbackTyperGroup(HFCliTyperGroup):533 def resolve_command(self, ctx: click.Context, args: list[str]) -> tuple:534 fallback_exit_code = fallback_handler(args, set(self.commands.keys()))535 if fallback_exit_code is not None:536 raise SystemExit(fallback_exit_code)537 return super().resolve_command(ctx, args)538 539 def format_commands(self, ctx: click.Context, formatter: click.HelpFormatter) -> None:540 super().format_commands(ctx, formatter)541 if extra_commands_provider is not None:542 entries = extra_commands_provider()543 if entries:544 with formatter.section("Extension commands"):545 formatter.write_dl(entries)546 547 return FallbackTyperGroup548 549 550def HFCliCommand(topic: TOPIC_T, examples: list[str] | None = None) -> type[TyperCommand]:551 def format_epilog(self: click.Command, ctx: click.Context, formatter: click.HelpFormatter) -> None:552 _format_epilog_no_indent(self.epilog, ctx, formatter)553 554 def format_options(self: TyperCommand, ctx: click.Context, formatter: click.HelpFormatter) -> None:555 TyperCommand.format_options(self, ctx, formatter)556 # Skip the section for commands that define their own --format / --quiet / --json,557 # or for pass-through commands that forward args to an external binary.558 if _has_local_formatting_option(self):559 return560 if self.context_settings.get("ignore_unknown_options"):561 return562 _format_formatting_options_section(formatter)563 564 def parse_args(self: click.Command, ctx: click.Context, args: list[str]) -> list[str]:565 # Show help when a command with required arguments is invoked without any args566 # (mirrors group behavior: `hf jobs` prints help, so `hf download` should too).567 if not args and not ctx.resilient_parsing:568 if any(isinstance(p, click.Argument) and p.required for p in self.params):569 click.echo(ctx.get_help(), color=ctx.color)570 ctx.exit()571 return TyperCommand.parse_args(self, ctx, args)572 573 return type(574 f"TyperCommand{topic.capitalize()}",575 (TyperCommand,),576 {577 "topic": topic,578 "examples": examples or [],579 "format_epilog": format_epilog,580 "format_options": format_options,581 "parse_args": parse_args,582 },583 )584 585 586class HFCliApp(typer.Typer):587 """Custom Typer app for Hugging Face CLI."""588 589 def command( # type: ignore590 self,591 name: str | None = None,592 *,593 topic: TOPIC_T = "main",594 examples: list[str] | None = None,595 context_settings: dict[str, Any] | None = None,596 help: str | None = None,597 epilog: str | None = None,598 short_help: str | None = None,599 options_metavar: str = "[OPTIONS]",600 add_help_option: bool = True,601 no_args_is_help: bool = False,602 hidden: bool = False,603 deprecated: bool = False,604 rich_help_panel: str | None = None,605 ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:606 # Generate epilog from examples if not explicitly provided607 if epilog is None and examples:608 epilog = generate_epilog(examples)609 610 def _inner(func: Callable[..., Any]) -> Callable[..., Any]:611 return super(HFCliApp, self).command(612 name,613 cls=HFCliCommand(topic, examples),614 context_settings=context_settings,615 help=help,616 epilog=epilog,617 short_help=short_help,618 options_metavar=options_metavar,619 add_help_option=add_help_option,620 no_args_is_help=no_args_is_help,621 hidden=hidden,622 deprecated=deprecated,623 rich_help_panel=rich_help_panel,624 )(func)625 626 return _inner627 628 629def typer_factory(help: str, epilog: str | None = None, cls: type[TyperGroup] | None = None) -> "HFCliApp":630 """Create a Typer app with consistent settings.631 632 Args:633 help: Help text for the app.634 epilog: Optional epilog text (use `generate_epilog` to create one).635 cls: Optional Click group class to use (defaults to `HFCliTyperGroup`).636 637 Returns:638 A configured Typer app.639 """640 if cls is None:641 cls = HFCliTyperGroup642 return HFCliApp(643 help=help,644 epilog=epilog,645 add_completion=True,646 no_args_is_help=True,647 cls=cls,648 # Disable rich completely for consistent experience649 rich_markup_mode=None,650 rich_help_panel=None,651 pretty_exceptions_enable=False,652 # Disable TyperGroup's suggest_commands, it matches against raw aliased653 # keys ("list | ls") leaking pipe syntax into user-facing messages.654 # HFCliTyperGroup.resolve_command() handles suggestions with expanded names.655 suggest_commands=False,656 # Increase max content width for better readability657 context_settings={658 "max_content_width": 120,659 "help_option_names": ["-h", "--help"],660 },661 )662 663 664class RepoType(str, Enum):665 model = "model"666 dataset = "dataset"667 space = "space"668 669 670RepoIdArg = Annotated[671 str,672 typer.Argument(673 help="The ID of the repo (e.g. `username/repo-name` or `spaces/username/repo-name`).",674 ),675]676 677 678RepoTypeOpt = Annotated[679 RepoType,680 typer.Option(681 "--type",682 "--repo-type",683 help="The type of repository (model, dataset, or space).",684 ),685]686 687TokenOpt = Annotated[688 str | None,689 typer.Option(690 help="A User Access Token generated from https://huggingface.co/settings/tokens.",691 ),692]693 694PrivateOpt = Annotated[695 bool | None,696 typer.Option(697 help="Whether to create a private repo if repo doesn't exist on the Hub. Ignored if the repo already exists.",698 ),699]700 701RevisionOpt = Annotated[702 str | None,703 typer.Option(704 help="Git revision id which can be a branch name, a tag, or a commit hash.",705 ),706]707 708 709LimitOpt = Annotated[710 int,711 typer.Option(help="Limit the number of results."),712]713 714AuthorOpt = Annotated[715 str | None,716 typer.Option(help="Filter by author or organization."),717]718 719FilterOpt = Annotated[720 list[str] | None,721 typer.Option(help="Filter by tags (e.g. 'text-classification'). Can be used multiple times."),722]723 724SearchOpt = Annotated[725 str | None,726 typer.Option(help="Search query."),727]728 729 730# --- Env / Secrets shared options and parsing helpers (used by jobs, repos, etc.) ---731 732EnvOpt = Annotated[733 list[str] | None,734 typer.Option(735 "-e",736 "--env",737 help="Set environment variables. E.g. --env ENV=value",738 ),739]740 741SecretsOpt = Annotated[742 list[str] | None,743 typer.Option(744 "-s",745 "--secrets",746 help=(747 "Set secret environment variables. E.g. --secrets SECRET=value"748 " or `--secrets HF_TOKEN` to pass your Hugging Face token."749 ),750 ),751]752 753EnvFileOpt = Annotated[754 str | None,755 typer.Option(756 "--env-file",757 help="Read in a file of environment variables.",758 ),759]760 761SecretsFileOpt = Annotated[762 str | None,763 typer.Option(764 help="Read in a file of secret environment variables.",765 ),766]767 768 769def _get_extended_environ() -> dict[str, str]:770 """Return a copy of ``os.environ`` with the user's HF token injected (if available)."""771 from huggingface_hub import get_token772 773 extended_environ = os.environ.copy()774 if (token := get_token()) is not None:775 extended_environ["HF_TOKEN"] = token776 return extended_environ777 778 779def parse_env_map(780 env: list[str] | None = None,781 env_file: str | None = None,782) -> dict[str, str | None]:783 """Parse ``-e``/``--env``/``-s``/``--secrets`` and ``--env-file``/``--secrets-file`` CLI args into a dict.784 785 Uses an extended environment that includes the user's HF token so that786 bare ``--secrets HF_TOKEN`` resolves correctly.787 """788 extended_environ = _get_extended_environ()789 env_map: dict[str, str | None] = {}790 if env_file:791 env_map.update(load_dotenv(Path(env_file).read_text(), environ=extended_environ))792 for env_value in env or []:793 env_map.update(load_dotenv(env_value, environ=extended_environ))794 return env_map795 796 797def env_map_to_key_value_list(env_map: dict[str, str | None]) -> list[dict[str, str]] | None:798 """Convert an env/secrets dict to the ``[{"key": ..., "value": ...}]`` format used by the Hub API."""799 if not env_map:800 return None801 return [{"key": k, "value": v or ""} for k, v in env_map.items()]802 803 804VolumesOpt = Annotated[805 list[str] | None,806 typer.Option(807 "-v",808 "--volume",809 help="Mount one or more volumes. Format: hf://[TYPE/]SOURCE:/MOUNT_PATH[:ro]. "810 "TYPE is one of: models, datasets, spaces, buckets. "811 "TYPE defaults to models if omitted. "812 "models, datasets and spaces are always mounted read-only. buckets are read+write by default. "813 "E.g. -v hf://org/m:/data or -v hf://datasets/org/ds:/data or -v hf://buckets/org/b:/mnt:ro",814 ),815]816 817_HF_PREFIX = "hf://"818_HF_VOLUME_TYPES = {819 "models": constants.REPO_TYPE_MODEL,820 "datasets": constants.REPO_TYPE_DATASET,821 "spaces": constants.REPO_TYPE_SPACE,822 "buckets": "bucket",823}824 825 826def parse_volumes(volumes: list[str] | None) -> "list[Volume] | None":827 """Parse volume specs from CLI arguments.828 829 Format: hf://[TYPE/]SOURCE[/PATH]:/MOUNT_PATH[:ro|:rw]830 Where TYPE is one of: models, datasets, spaces, buckets (defaults to models if omitted).831 SOURCE is the repo/bucket identifier (e.g. 'username/my-model').832 PATH is an optional subfolder inside the repo/bucket.833 MOUNT_PATH starts with '/'.834 Optional ':ro' or ':rw' suffix for read-only or read-write.835 836 Examples:837 hf://my-org/my-model:/data (model, implicit type)838 hf://models/my-org/my-model:/data (model, explicit type)839 hf://datasets/my-org/my-dataset:/data:ro840 hf://buckets/my-org/my-bucket:/mnt841 hf://spaces/my-org/my-space:/app842 hf://datasets/org/ds/train:/data (with path inside repo)843 hf://buckets/org/b/sub/dir:/mnt (with path inside bucket)844 """845 846 if not volumes:847 return None848 849 result: list[Volume] = []850 for raw_spec in volumes:851 # Strip :ro/:rw suffix852 spec = raw_spec853 read_only = None854 if spec.endswith(":ro"):855 read_only = True856 spec = spec[:-3]857 elif spec.endswith(":rw"):858 read_only = False859 spec = spec[:-3]860 861 # Validate hf:// prefix862 if not spec.startswith(_HF_PREFIX):863 raise CLIError(864 f"Invalid volume format: '{raw_spec}'. Source must start with 'hf://'. "865 f"Expected hf://[TYPE/]SOURCE:/MOUNT_PATH[:ro]. E.g. hf://org/m:/data"866 )867 spec = spec[len(_HF_PREFIX) :]868 869 # Find the mount path: look for :/ pattern870 colon_slash_idx = spec.find(":/")871 if colon_slash_idx == -1:872 raise CLIError(873 f"Invalid volume format: '{raw_spec}'. Expected hf://[TYPE/]SOURCE:/MOUNT_PATH[:ro]. E.g. hf://org/m:/data"874 )875 source_part = spec[:colon_slash_idx]876 mount_path = spec[colon_slash_idx + 1 :]877 878 # Parse type from source_part (first segment before /)879 # Then split remaining into source (namespace/name or name) and optional path.880 slash_idx = source_part.find("/")881 if slash_idx == -1:882 # No slash: bare source like "gpt2" -> model type883 vol_type_str = constants.REPO_TYPE_MODEL884 source = source_part885 path = None886 else:887 first_segment = source_part[:slash_idx]888 if first_segment in _HF_VOLUME_TYPES:889 vol_type_str = _HF_VOLUME_TYPES[first_segment]890 remaining = source_part[slash_idx + 1 :]891 else:892 # First segment isn't a known type -> model type893 vol_type_str = constants.REPO_TYPE_MODEL894 remaining = source_part895 896 # Split remaining into source (namespace/name) and optional path.897 # Repo/bucket IDs are "namespace/name" (2 segments) or "name" (1 segment).898 # Any extra segments are the path inside the repo/bucket.899 parts = remaining.split("/", 2)900 if len(parts) >= 3:901 source = parts[0] + "/" + parts[1]902 path = parts[2]903 else:904 source = remaining905 path = None906 907 result.append(908 Volume(909 type=vol_type_str,910 source=source,911 mount_path=mount_path,912 read_only=read_only,913 path=path,914 )915 )916 return result917 918 919class OutputFormat(str, Enum):920 """Output format for CLI list commands."""921 922 table = "table"923 json = "json"924 925 926FormatOpt = Annotated[927 OutputFormat,928 typer.Option(929 help="Output format (table or json).",930 ),931]932 933 934def _set_output_mode(value: OutputFormatWithAuto) -> OutputFormatWithAuto:935 """Callback for the legacy FormatWithAutoOpt option type.936 937 Most commands now rely on the global --format / --json / -q flags consumed by _consume_format_flags_for_leaf instead938 of declaring FormatWithAutoOpt themselves. This callback is kept for the rare cases where a command still wires939 FormatWithAutoOpt explicitly.940 """941 out.set_mode(value)942 return value943 944 945FormatWithAutoOpt = Annotated[946 OutputFormatWithAuto,947 typer.Option(help="Output format.", callback=_set_output_mode),948]949 950QuietOpt = Annotated[951 bool,952 typer.Option("-q", "--quiet", help="Print only IDs (one per line)."),953]954 955 956def _to_header(name: str) -> str:957 """Convert a camelCase or PascalCase string to SCREAMING_SNAKE_CASE to be used as table header."""958 s = re.sub(r"([a-z])([A-Z])", r"\1_\2", name)959 return s.upper()960 961 962def _format_value(value: Any) -> str:963 """Convert a value to string for terminal display."""964 if not value:965 return ""966 if isinstance(value, bool):967 return "✔" if value else ""968 if isinstance(value, datetime.datetime):969 return value.strftime("%Y-%m-%d")970 if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2}T", value):971 return value[:10]972 if isinstance(value, list):973 return ", ".join(_format_value(v) for v in value)974 elif isinstance(value, dict):975 if "name" in value: # Likely to be a user or org => print name976 return str(value["name"])977 # TODO: extend if needed978 return json.dumps(value)979 return str(value)980 981 982def _format_cell(value: Any, max_len: int = _MAX_CELL_LENGTH) -> str:983 """Format a value + truncate it for table display."""984 cell = _format_value(value)985 if len(cell) > max_len:986 cell = cell[: max_len - 3] + "..."987 return cell988 989 990def print_as_table(991 items: Sequence[dict[str, Any]],992 headers: list[str],993 row_fn: Callable[[dict[str, Any]], list[str]],994 alignments: dict[str, str] | None = None,995) -> None:996 """Print items as a formatted table.997 998 Args:999 items: Sequence of dictionaries representing the items to display.1000 headers: List of column headers.1001 row_fn: Function that takes an item dict and returns a list of string values for each column.1002 alignments: Optional mapping of header name to "left" or "right". Defaults to "left".1003 """1004 if not items:1005 print("No results found.")1006 return1007 rows = cast(list[list[Union[str, int]]], [row_fn(item) for item in items])1008 screaming_headers = [_to_header(h) for h in headers]1009 # Remap alignments keys to screaming case to match tabulate headers1010 screaming_alignments = {_to_header(k): v for k, v in (alignments or {}).items()}1011 print(tabulate(rows, headers=screaming_headers, alignments=screaming_alignments))1012 1013 1014def print_list_output(1015 items: Sequence[dict[str, Any]],1016 format: OutputFormat,1017 quiet: bool,1018 id_key: str = "id",1019 headers: list[str] | None = None,1020 row_fn: Callable[[dict[str, Any]], list[str]] | None = None,1021 alignments: dict[str, str] | None = None,1022) -> None:1023 """Print list command output in the specified format.1024 1025 Args:1026 items: Sequence of dictionaries representing the items to display.1027 format: Output format.1028 quiet: If True, print only IDs (one per line).1029 id_key: Key to use for extracting IDs in quiet mode.1030 headers: Optional list of column names for headers. If not provided, auto-detected from keys.1031 row_fn: Optional function to extract row values. If not provided, uses _format_cell on each column.1032 alignments: Optional mapping of header name to "left" or "right". Defaults to "left".1033 """1034 if quiet:1035 for item in items:1036 print(item[id_key])1037 return1038 1039 if format == OutputFormat.json:1040 print(json.dumps(list(items), indent=2, default=str))1041 return1042 1043 if headers is None:1044 all_columns = list(items[0].keys()) if items else [id_key]1045 headers = [col for col in all_columns if any(_format_cell(item.get(col)) for item in items)]1046 1047 if row_fn is None:1048 1049 def row_fn(item: dict[str, Any]) -> list[str]:1050 return [_format_cell(item.get(col)) for col in headers] # type: ignore[union-attr]1051 1052 print_as_table(items, headers=headers, row_fn=row_fn, alignments=alignments)1053 1054 1055def _serialize_value(v: object) -> object:1056 """Recursively serialize a value to be JSON-compatible."""1057 if isinstance(v, datetime.datetime):1058 return v.isoformat()1059 elif isinstance(v, dict):1060 return {key: _serialize_value(val) for key, val in v.items() if val is not None}1061 elif isinstance(v, list):1062 return [_serialize_value(item) for item in v]1063 return v1064 1065 1066def api_object_to_dict(info: Any) -> dict[str, Any]:1067 """Convert repo info dataclasses to json-serializable dicts."""1068 return {k: _serialize_value(v) for k, v in dataclasses.asdict(info).items() if v is not None}1069 1070 1071def make_expand_properties_parser(valid_properties: Sequence[ExpandPropertyT]):1072 """Create a callback to parse and validate comma-separated expand properties."""1073 1074 def _parse_expand_properties(value: str | None) -> list[ExpandPropertyT] | None:1075 if value is None:1076 return None1077 properties = [p.strip() for p in value.split(",")]1078 for prop in properties:1079 if prop not in valid_properties:1080 raise typer.BadParameter(1081 f"Invalid expand property: '{prop}'. Valid values are: {', '.join(valid_properties)}"1082 )1083 return [cast(ExpandPropertyT, prop) for prop in properties]1084 1085 return _parse_expand_properties1086 1087 1088### PyPI VERSION CHECKER1089 1090 1091def check_cli_update(library: Literal["huggingface_hub", "transformers"]) -> None:1092 """1093 Check whether a newer version of a library is available on PyPI.1094 1095 If a newer version is found, print a hint pointing at `hf update`.1096 1097 If current version is a pre-release (e.g. `1.0.0.rc1`), or a dev version (e.g. `1.0.0.dev1`), no check is performed.1098 If `HF_HUB_DISABLE_UPDATE_CHECK` is set, the check is skipped entirely.1099 1100 This function is called at the entry point of the CLI. It only performs the check once every 24 hours, and any error1101 during the check is caught and logged, to avoid breaking the CLI.1102 1103 Args:1104 library: The library to check for updates. Currently supports "huggingface_hub" and "transformers".1105 """1106 try:1107 _check_cli_update(library)1108 except Exception:1109 # We don't want the CLI to fail on version checks, no matter the reason.1110 logger.debug("Error while checking for CLI update.", exc_info=True)1111 1112 1113def _check_cli_update(library: Literal["huggingface_hub", "transformers"]) -> None:1114 if constants.HF_HUB_DISABLE_UPDATE_CHECK:1115 return1116 1117 current_version = importlib.metadata.version(library)1118 1119 # Skip if current version is a pre-release or dev version1120 if any(tag in current_version for tag in ["rc", "dev"]):1121 return1122 1123 # Skip if already checked in the last 24 hours1124 if os.path.exists(constants.CHECK_FOR_UPDATE_DONE_PATH):1125 mtime = os.path.getmtime(constants.CHECK_FOR_UPDATE_DONE_PATH)1126 if (time.time() - mtime) < 24 * 3600:1127 return1128 1129 # Touch the file to mark that we did the check now1130 Path(constants.CHECK_FOR_UPDATE_DONE_PATH).parent.mkdir(parents=True, exist_ok=True)1131 Path(constants.CHECK_FOR_UPDATE_DONE_PATH).touch()1132 1133 # Check latest version from PyPI1134 latest_version = _fetch_latest_pypi_version(library)1135 if latest_version is None or current_version == latest_version:1136 return1137 1138 if library == "huggingface_hub":1139 update_command = _get_huggingface_hub_update_command()1140 else:1141 update_command = _get_transformers_update_command()1142 1143 message = f"A new version of {library} ({latest_version}) is available! You are using version {current_version}."1144 if update_command is not None:1145 match library:1146 case "huggingface_hub":1147 message += "\nTo update, run: hf update"1148 case _:1149 message += f"\nTo update, run: {' '.join(update_command)}"1150 out.hint(message)1151 1152 1153def _fetch_latest_pypi_version(library: str) -> str | None:1154 """Fetch the latest version of a library from PyPI. Returns None if the request fails."""1155 try:1156 response = get_session().get(f"https://pypi.org/pypi/{library}/json", timeout=2)1157 hf_raise_for_status(response)1158 return response.json()["info"]["version"]1159 except Exception:1160 logger.debug("Error while fetching latest version from PyPI.", exc_info=True)1161 return None1162 1163 1164def run_update() -> int:1165 """Run the install-method-appropriate update command for the `hf` CLI.1166 1167 Raises CLIError if the installation method can't be determined.1168 Returns the subprocess exit code on success/failure of the update itself.1169 """1170 cmd = _get_huggingface_hub_update_command()1171 if cmd is None:1172 raise CLIError(1173 "Cannot determine how to update huggingface_hub (unknown installation method). Please update manually."1174 )1175 return subprocess.call(cmd)1176 1177 1178def _get_huggingface_hub_update_command() -> list[str] | None:1179 """Return the command to update huggingface_hub as an argv list, or None if the installation method is unknown."""1180 match installation_method():1181 case "brew":1182 return ["brew", "upgrade", "hf"]1183 case "hf_installer" if os.name == "nt":1184 return ["powershell", "-NoProfile", "-Command", "iwr -useb https://hf.co/cli/install.ps1 | iex"]1185 case "hf_installer":1186 return ["bash", "-c", "curl -LsSf https://hf.co/cli/install.sh | bash -"]1187 case "pip":1188 return [sys.executable, "-m", "pip", "install", "-U", "huggingface_hub"]1189 case _:1190 return None1191 1192 1193def _get_transformers_update_command() -> list[str] | None:1194 """Return the command to update transformers as an argv list, or None if the installation method is unknown."""1195 match installation_method():1196 case "hf_installer" if os.name == "nt":1197 return [1198 "powershell",1199 "-NoProfile",1200 "-Command",