1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import importlib.util
17import io
18import os
19import tempfile
20from collections.abc import Callable, Iterator
21from typing import IO, Any, TypeVar
22
23from huggingface_hub import CommitOperationCopy, HfApi
24from huggingface_hub.errors import EntryNotFoundError, HfHubHTTPError, RepositoryNotFoundError, RevisionNotFoundError
25from huggingface_hub.hf_api import RepoFile, RepoFolder
26
27from ..telemetry import Telemetry
28from ..types import (
29 AWARE_DATETIME_MIN,
30 Credentials,
31 CredentialsProvider,
32 ObjectMetadata,
33 Range,
34 RetryableError,
35 SymlinkHandling,
36)
37from ..utils import safe_makedirs
38from .base import BaseStorageProvider
39
40_T = TypeVar("_T")
41
42PROVIDER = "huggingface"
43
44HF_TRANSFER_UNAVAILABLE_ERROR_MESSAGE = (
45 "Fast transfer using 'hf_transfer' is enabled (HF_HUB_ENABLE_HF_TRANSFER=1) "
46 "but 'hf_transfer' package is not available in your environment. "
47 "Either install hf_transfer with 'pip install hf_transfer' or "
48 "disable it by setting HF_HUB_ENABLE_HF_TRANSFER=0"
49)
50
51
[docs]
52class HuggingFaceCredentialsProvider(CredentialsProvider):
53 """
54 A concrete implementation of the :py:class:`multistorageclient.types.CredentialsProvider` that provides HuggingFace credentials.
55 """
56
57 def __init__(self, access_token: str):
58 """
59 Initializes the :py:class:`HuggingFaceCredentialsProvider` with the provided access token.
60
61 :param access_token: The HuggingFace access token for authentication.
62 """
63 self.token = access_token
64
[docs]
65 def get_credentials(self) -> Credentials:
66 """
67 Retrieves the current HuggingFace credentials.
68
69 :return: The current credentials used for HuggingFace authentication.
70 """
71 return Credentials(
72 access_key="",
73 secret_key="",
74 token=self.token,
75 expiration=None,
76 )
77
[docs]
78 def refresh_credentials(self) -> None:
79 """
80 Refreshes the credentials if they are expired or about to expire.
81
82 Note: HuggingFace tokens typically don't expire, so this is a no-op.
83 """
84
85
[docs]
86class HuggingFaceStorageProvider(BaseStorageProvider):
87 """
88 A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with HuggingFace Hub repositories.
89 """
90
91 def __init__(
92 self,
93 repository_id: str,
94 repo_type: str = "model",
95 base_path: str = "",
96 repo_revision: str = "main",
97 credentials_provider: CredentialsProvider | None = None,
98 config_dict: dict[str, Any] | None = None,
99 telemetry_provider: Callable[[], Telemetry] | None = None,
100 ):
101 """
102 Initializes the :py:class:`HuggingFaceStorageProvider` with repository information and optional credentials provider.
103
104 :param repository_id: The HuggingFace repository ID (e.g., 'username/repo-name').
105 :param repo_type: The type of repository ('dataset', 'model', 'space'). Defaults to 'model'.
106 :param base_path: The root prefix path within the repository where all operations will be scoped.
107 :param repo_revision: The git revision (branch, tag, or commit) to use. Defaults to 'main'.
108 :param credentials_provider: The provider to retrieve HuggingFace credentials.
109 :param config_dict: Resolved MSC config.
110 :param telemetry_provider: A function that provides a telemetry instance.
111 """
112
113 # Validate repo_type
114 allowed_repo_types = {"dataset", "model", "space"}
115 if repo_type not in allowed_repo_types:
116 raise ValueError(f"Invalid repo_type '{repo_type}'. Must be one of: {allowed_repo_types}")
117
118 # Validate repository_id format
119 if not repository_id or "/" not in repository_id:
120 raise ValueError(f"Invalid repository_id '{repository_id}'. Expected format: 'username/repo-name'")
121
122 self._validate_hf_transfer_availability()
123
124 super().__init__(
125 base_path=base_path,
126 provider_name=PROVIDER,
127 config_dict=config_dict,
128 telemetry_provider=telemetry_provider,
129 )
130
131 self._repository_id = repository_id
132 self._repo_type = repo_type
133 self._repo_revision = repo_revision
134 self._credentials_provider = credentials_provider
135
136 self._hf_client: HfApi = self._create_hf_api_client()
137
138 def _create_hf_api_client(self) -> HfApi:
139 """
140 Creates and configures the HuggingFace API client.
141
142 Initializes the HfApi client with authentication token if credentials are provided,
143 otherwise creates an unauthenticated client for public repositories.
144
145 :return: Configured HfApi client instance.
146 """
147
148 token = None
149 if self._credentials_provider:
150 creds = self._credentials_provider.get_credentials()
151 token = creds.token
152
153 return HfApi(token=token)
154
155 def _validate_hf_transfer_availability(self) -> None:
156 """
157 Validates that hf_transfer is available if it's enabled via environment variables.
158
159 Raises:
160 ValueError: If hf_transfer is enabled but not available.
161 """
162 # Check if hf_transfer is enabled via environment variable
163 hf_transfer_enabled = os.environ.get("HF_HUB_ENABLE_HF_TRANSFER", "").lower() in ("1", "on", "true", "yes")
164
165 if hf_transfer_enabled and importlib.util.find_spec("hf_transfer") is None:
166 raise ValueError(HF_TRANSFER_UNAVAILABLE_ERROR_MESSAGE)
167
168 def _parse_rate_limit_headers(self, response) -> str:
169 """
170 Parses HuggingFace rate limit headers and returns formatted information.
171
172 HuggingFace returns rate limit information in these headers:
173 - RateLimit: "api";r=0;t=142
174 - r = requests remaining in the current window
175 - t = seconds until rate limit resets
176 - RateLimit-Policy: "fixed window";"api";q=10000;w=300
177 - q = total requests allowed per window
178 - w = window size in seconds
179
180 Reference: https://huggingface.co/docs/hub/rate-limits
181
182 :param response: The HTTP response object containing rate limit headers.
183 :return: Formatted string with rate limit information, or empty string if headers not found.
184 """
185
186 try:
187 headers = response.headers
188 except Exception:
189 return ""
190
191 rate_limit_info = []
192
193 # Note: HTTP headers are case-insensitive, but we use the canonical casing from HF docs
194 if "RateLimit" in headers:
195 rate_limit = headers["RateLimit"]
196 # Extract r (remaining) and t (time until reset)
197 remaining = None
198 reset_seconds = None
199
200 parts = rate_limit.split(";")
201 for part in parts:
202 part = part.strip()
203 if part.startswith("r="):
204 try:
205 remaining = int(part[2:])
206 except ValueError:
207 pass
208 elif part.startswith("t="):
209 try:
210 reset_seconds = int(part[2:])
211 except ValueError:
212 pass
213
214 if remaining is not None:
215 rate_limit_info.append(f"Requests remaining in current window: {remaining}")
216 if reset_seconds is not None:
217 rate_limit_info.append(f"Rate limit resets in: {reset_seconds} seconds")
218
219 if "RateLimit-Policy" in headers:
220 policy = headers["RateLimit-Policy"]
221 # Extract q (quota) and w (window size)
222 quota = None
223 window_seconds = None
224
225 parts = policy.split(";")
226 for part in parts:
227 part = part.strip()
228 if part.startswith("q="):
229 try:
230 quota = int(part[2:])
231 except ValueError:
232 pass
233 elif part.startswith("w="):
234 try:
235 window_seconds = int(part[2:])
236 except ValueError:
237 pass
238
239 if quota is not None and window_seconds is not None:
240 window_minutes = window_seconds / 60
241 rate_limit_info.append(f"Rate limit policy: {quota} requests per {window_minutes:.0f}-minute window")
242
243 if rate_limit_info:
244 return " | ".join(rate_limit_info)
245
246 return ""
247
248 def _translate_errors(
249 self,
250 func: Callable[[], _T],
251 operation: str,
252 repo_id: str,
253 path: str,
254 ) -> _T:
255 """
256 Translates HuggingFace errors into standardized exceptions with retry logic.
257
258 Parses HuggingFace rate limit headers (RateLimit and RateLimit-Policy) to provide
259 detailed information about rate limiting to users. See https://huggingface.co/docs/hub/rate-limits
260
261 :param func: The function that performs the actual HuggingFace operation.
262 :param operation: The type of operation being performed (e.g., "upload", "download", "delete").
263 :param repo_id: The HuggingFace repository ID.
264 :param path: The path of the object within the repository.
265 :return: The result of the HuggingFace operation.
266 :raises RetryableError: For transient errors that can be retried (429, 503, connection errors).
267 :raises FileNotFoundError: When the requested resource is not found.
268 :raises RuntimeError: For other non-retryable errors.
269 """
270 try:
271 return func()
272 except RepositoryNotFoundError as error:
273 raise FileNotFoundError(
274 f"Repository not found or access denied: {repo_id}. "
275 f"Verify the repository exists and you have access permissions."
276 ) from error
277 except RevisionNotFoundError as error:
278 raise FileNotFoundError(
279 f"Revision '{self._repo_revision}' not found in repository {repo_id}. "
280 f"Verify the branch, tag, or commit exists."
281 ) from error
282 except EntryNotFoundError as error:
283 raise FileNotFoundError(f"File not found in HuggingFace repository: {path}") from error
284 except FileNotFoundError:
285 raise
286 except HfHubHTTPError as error:
287 # Extract status code and parse rate limit headers
288 # Don't use hasattr() - it's unreliable with response objects
289 status_code = None
290 response = None
291
292 try:
293 response = error.response
294 if response is not None:
295 status_code = response.status_code
296 except AttributeError:
297 pass
298
299 rate_limit_info = self._parse_rate_limit_headers(response)
300 quota_suffix = f" | {rate_limit_info}" if rate_limit_info else ""
301
302 error_info = f"repo_id: {repo_id}, path: {path}, status_code: {status_code}, error: {error}"
303
304 if status_code == 404:
305 raise FileNotFoundError(f"Object {repo_id}/{path} does not exist. {error_info}") from error
306 elif status_code == 409:
307 raise RetryableError(f"Conflict Error for {repo_id}. {error_info}{quota_suffix}") from error
308 elif status_code == 429:
309 base_message = f"Rate limit exceeded when {operation} object(s) at {repo_id}/{path}. {error_info}"
310 raise RetryableError(f"{base_message}{quota_suffix}") from error
311 elif status_code == 503:
312 raise RetryableError(
313 f"Service unavailable when {operation} object(s) at {repo_id}/{path}. {error_info}{quota_suffix}"
314 ) from error
315 elif status_code in (408, 500, 502, 504):
316 raise RetryableError(
317 f"Transient error ({status_code}) when {operation} object(s) at {repo_id}/{path}. {error_info}{quota_suffix}"
318 ) from error
319 else:
320 raise RuntimeError(
321 f"HuggingFace API error during {operation} of {path}: {error}{quota_suffix}"
322 ) from error
323 except (ConnectionError, TimeoutError, OSError) as error:
324 raise RetryableError(
325 f"Connection error when {operation} object(s) at {repo_id}/{path}, error type: {type(error).__name__}"
326 ) from error
327 except Exception as error:
328 raise RuntimeError(f"Unexpected error during {operation} of {path}: {error}") from error
329
330 def _put_object(
331 self,
332 path: str,
333 body: bytes,
334 if_match: str | None = None,
335 if_none_match: str | None = None,
336 attributes: dict[str, str] | None = None,
337 ) -> int:
338 """
339 Uploads an object to the HuggingFace repository.
340
341 :param path: The path where the object will be stored in the repository.
342 :param body: The content of the object to store.
343 :param if_match: Optional ETag for conditional uploads (not supported by HuggingFace).
344 :param if_none_match: Optional ETag for conditional uploads (not supported by HuggingFace).
345 :param attributes: Optional attributes for the object (not supported by HuggingFace).
346 :return: Data size in bytes.
347 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
348 :raises ValueError: If client attempts to create a directory.
349 :raises ValueError: If conditional upload parameters are provided (not supported).
350 """
351 if not self._hf_client:
352 raise RuntimeError("HuggingFace client not initialized")
353
354 if if_match is not None or if_none_match is not None:
355 raise ValueError(
356 "HuggingFace provider does not support conditional uploads. "
357 "if_match and if_none_match parameters are not supported."
358 )
359
360 if attributes is not None:
361 raise ValueError(
362 "HuggingFace provider does not support custom object attributes. "
363 "Use commit messages or repository metadata instead."
364 )
365
366 if path.endswith("/"):
367 raise ValueError(
368 "HuggingFace Storage Provider does not support explicit directory creation. "
369 "Directories are created implicitly when files are uploaded to paths within them."
370 )
371
372 path = self._normalize_path(path)
373
374 def _invoke_api():
375 with tempfile.NamedTemporaryFile(delete=False) as temp_file:
376 temp_file.write(body)
377 temp_file_path = temp_file.name
378
379 try:
380 self._hf_client.upload_file(
381 path_or_fileobj=temp_file_path,
382 path_in_repo=path,
383 repo_id=self._repository_id,
384 repo_type=self._repo_type,
385 revision=self._repo_revision,
386 commit_message=f"Upload {path}",
387 commit_description=None,
388 create_pr=False,
389 )
390
391 return len(body)
392
393 finally:
394 os.unlink(temp_file_path)
395
396 return self._translate_errors(_invoke_api, "PUT", self._repository_id, path)
397
398 def _get_object(self, path: str, byte_range: Range | None = None) -> bytes:
399 """
400 Retrieves an object from the HuggingFace repository.
401
402 :param path: The path of the object to retrieve from the repository.
403 :param byte_range: Optional byte range for partial content (not supported by HuggingFace).
404 :return: The content of the retrieved object.
405 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
406 :raises ValueError: If a byte range is requested (HuggingFace doesn't support range reads).
407 :raises FileNotFoundError: If the file doesn't exist in the repository.
408 """
409
410 if not self._hf_client:
411 raise RuntimeError("HuggingFace client not initialized")
412
413 if byte_range is not None:
414 raise ValueError(
415 "HuggingFace provider does not support partial range reads. "
416 f"Requested range: offset={byte_range.offset}, size={byte_range.size}. "
417 "To read the entire file, call get_object() without the byte_range parameter."
418 )
419
420 path = self._normalize_path(path)
421
422 def _invoke_api():
423 with tempfile.TemporaryDirectory() as temp_dir:
424 downloaded_path = self._hf_client.hf_hub_download(
425 repo_id=self._repository_id,
426 filename=path,
427 repo_type=self._repo_type,
428 revision=self._repo_revision,
429 local_dir=temp_dir,
430 )
431
432 with open(downloaded_path, "rb") as f:
433 data = f.read()
434
435 return data
436
437 return self._translate_errors(_invoke_api, "GET", self._repository_id, path)
438
439 def _copy_object(self, src_path: str, dest_path: str) -> int:
440 """
441 Copies an object within the HuggingFace repository using server-side copy.
442
443 .. note::
444 Copy behavior is size-dependent: files ≥10MB are copied remotely via
445 metadata (LFS), while files <10MB are downloaded and re-uploaded.
446
447 :param src_path: The source path of the object to copy.
448 :param dest_path: The destination path for the copied object.
449 :return: Data size in bytes.
450 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
451 :raises FileNotFoundError: If the source file doesn't exist.
452 """
453 if not self._hf_client:
454 raise RuntimeError("HuggingFace client not initialized")
455
456 src_path = self._normalize_path(src_path)
457 dest_path = self._normalize_path(dest_path)
458
459 src_object = self._get_object_metadata(src_path)
460
461 def _invoke_api():
462 operations = [
463 CommitOperationCopy(
464 src_path_in_repo=src_path,
465 path_in_repo=dest_path,
466 )
467 ]
468
469 self._hf_client.create_commit(
470 repo_id=self._repository_id,
471 operations=operations,
472 commit_message=f"Copy {src_path} to {dest_path}",
473 repo_type=self._repo_type,
474 revision=self._repo_revision,
475 )
476
477 return src_object.content_length
478
479 return self._translate_errors(_invoke_api, "COPY", self._repository_id, f"{src_path} to {dest_path}")
480
481 def _delete_object(self, path: str, if_match: str | None = None) -> None:
482 """
483 Deletes an object from the HuggingFace repository.
484
485 :param path: The path of the object to delete from the repository.
486 :param if_match: Optional ETag for conditional deletion (not supported by HuggingFace).
487 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
488 :raises ValueError: If conditional deletion parameters are provided (not supported).
489 :raises FileNotFoundError: If the file doesn't exist in the repository.
490 """
491 if not self._hf_client:
492 raise RuntimeError("HuggingFace client not initialized")
493
494 if if_match is not None:
495 raise ValueError(
496 "HuggingFace provider does not support conditional deletion. if_match parameter is not supported."
497 )
498
499 path = self._normalize_path(path)
500
501 def _invoke_api():
502 self._hf_client.delete_file(
503 path_in_repo=path,
504 repo_id=self._repository_id,
505 repo_type=self._repo_type,
506 revision=self._repo_revision,
507 commit_message=f"Delete {path}",
508 )
509
510 self._translate_errors(_invoke_api, "DELETE", self._repository_id, path)
511
512 def _item_to_metadata(self, item: RepoFile | RepoFolder) -> ObjectMetadata:
513 """
514 Convert a RepoFile or RepoFolder into ObjectMetadata.
515
516 :param item: The RepoFile or RepoFolder item from HuggingFace API.
517 :return: ObjectMetadata representing the item.
518 """
519 last_modified = AWARE_DATETIME_MIN
520
521 if isinstance(item, RepoFile):
522 etag = item.blob_id
523 return ObjectMetadata(
524 key=item.path,
525 type="file",
526 content_length=item.size,
527 last_modified=last_modified,
528 etag=etag,
529 content_type=None,
530 storage_class=None,
531 metadata=None,
532 )
533 else:
534 etag = item.tree_id
535 return ObjectMetadata(
536 key=item.path,
537 type="directory",
538 content_length=0,
539 last_modified=last_modified,
540 etag=etag,
541 content_type=None,
542 storage_class=None,
543 metadata=None,
544 )
545
546 def _make_symlink(self, path: str, target: str) -> None:
547 """
548 Not supported. HuggingFace repositories are read-only through this provider.
549
550 :raises NotImplementedError: Always.
551 """
552 raise NotImplementedError("HuggingFace provider does not support symlink creation.")
553
554 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
555 """
556 Retrieves metadata for an object in the HuggingFace repository.
557
558 :param path: The path of the object to get metadata for.
559 :param strict: Whether to raise an error if the object doesn't exist.
560 :return: Metadata about the object.
561 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
562 :raises FileNotFoundError: If the file doesn't exist and strict=True.
563 """
564 if not self._hf_client:
565 raise RuntimeError("HuggingFace client not initialized")
566
567 path = self._normalize_path(path)
568
569 def _invoke_api():
570 items = self._hf_client.get_paths_info(
571 repo_id=self._repository_id,
572 paths=[path],
573 repo_type=self._repo_type,
574 revision=self._repo_revision,
575 expand=True,
576 )
577
578 if not items:
579 raise FileNotFoundError(f"File not found in HuggingFace repository: {path}")
580
581 item = items[0]
582 return self._item_to_metadata(item)
583
584 try:
585 return self._translate_errors(_invoke_api, "HEAD", self._repository_id, path)
586 except FileNotFoundError:
587 if strict:
588 dir_path = path.rstrip("/") + "/"
589 if self._is_dir(dir_path):
590 return ObjectMetadata(
591 key=dir_path,
592 type="directory",
593 content_length=0,
594 last_modified=AWARE_DATETIME_MIN,
595 etag=None,
596 content_type=None,
597 storage_class=None,
598 metadata=None,
599 )
600 raise
601
602 def _list_objects(
603 self,
604 path: str,
605 start_after: str | None = None,
606 end_at: str | None = None,
607 include_directories: bool = False,
608 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
609 ) -> Iterator[ObjectMetadata]:
610 """
611 Lists objects in the HuggingFace repository under the specified path.
612
613 :param path: The path to list objects under.
614 :param start_after: The key to start listing after (exclusive, used as cursor).
615 :param end_at: The key to end listing at (inclusive, used as cursor).
616 :param include_directories: Whether to include directories in the listing.
617 :return: An iterator over object metadata for objects under the specified path.
618 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
619
620 .. note::
621 HuggingFace Hub API does not natively support pagination parameters.
622 This implementation fetches all items and uses cursor-based filtering,
623 which may impact performance for large repositories. The ordering is
624 directory-first, then files, with lexicographical ordering within each group.
625 """
626 if not self._hf_client:
627 raise RuntimeError("HuggingFace client not initialized")
628
629 path = self._normalize_path(path)
630
631 try:
632 metadata = self._get_object_metadata(path.rstrip("/"), strict=False)
633 if metadata and metadata.type == "file":
634 yield metadata
635 return
636 except FileNotFoundError:
637 pass
638
639 def _invoke_api():
640 dir_path = path.rstrip("/")
641
642 repo_items = self._hf_client.list_repo_tree(
643 repo_id=self._repository_id,
644 path_in_repo=dir_path + "/" if dir_path else None,
645 repo_type=self._repo_type,
646 revision=self._repo_revision,
647 expand=True,
648 recursive=not include_directories,
649 )
650
651 return list(repo_items)
652
653 try:
654 items = self._translate_errors(_invoke_api, "LIST", self._repository_id, path)
655
656 # Use cursor-based pagination because HuggingFace returns items with
657 # directory-first ordering (not pure lexicographical).
658 seen_start = start_after is None
659 seen_end = False
660
661 for item in items:
662 if seen_end:
663 break
664
665 metadata = self._item_to_metadata(item)
666 key = metadata.key
667
668 if not seen_start:
669 if key == start_after:
670 seen_start = True
671 continue
672
673 should_yield = False
674 if include_directories and isinstance(item, RepoFolder) or isinstance(item, RepoFile):
675 should_yield = True
676
677 if should_yield:
678 yield metadata
679
680 if end_at is not None and key == end_at:
681 seen_end = True
682
683 except FileNotFoundError:
684 # Directory doesn't exist - return empty (matches POSIX behavior)
685 pass
686
687 def _upload_file(self, remote_path: str, f: str | IO, attributes: dict[str, str] | None = None) -> int:
688 """
689 Uploads a file to the HuggingFace repository.
690
691 :param remote_path: The remote path where the file will be stored in the repository.
692 :param f: File path or file object to upload.
693 :param attributes: Optional attributes for the file (not supported by HuggingFace).
694 :return: Data size in bytes.
695 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
696 :raises ValueError: If client attempts to create a directory.
697 :raises ValueError: If custom attributes are provided (not supported).
698 """
699 if not self._hf_client:
700 raise RuntimeError("HuggingFace client not initialized")
701
702 if attributes is not None:
703 raise ValueError(
704 "HuggingFace provider does not support custom file attributes. "
705 "Use commit messages or repository metadata instead."
706 )
707
708 if remote_path.endswith("/"):
709 raise ValueError(
710 "HuggingFace Storage Provider does not support explicit directory creation. "
711 "Directories are created implicitly when files are uploaded to paths within them."
712 )
713
714 remote_path = self._normalize_path(remote_path)
715
716 def _invoke_api():
717 if isinstance(f, str):
718 file_size = os.path.getsize(f)
719
720 self._hf_client.upload_file(
721 path_or_fileobj=f,
722 path_in_repo=remote_path,
723 repo_id=self._repository_id,
724 repo_type=self._repo_type,
725 revision=self._repo_revision,
726 commit_message=f"Upload {remote_path}",
727 commit_description=None,
728 create_pr=False,
729 )
730
731 return file_size
732
733 else:
734 content = f.read()
735
736 if isinstance(content, str):
737 content_bytes = content.encode("utf-8")
738 else:
739 content_bytes = content
740
741 # Create temporary file since HfAPI.upload_file requires BinaryIO, not generic IO
742 with tempfile.NamedTemporaryFile(delete=False) as temp_file:
743 temp_file.write(content_bytes)
744 temp_file_path = temp_file.name
745
746 try:
747 self._hf_client.upload_file(
748 path_or_fileobj=temp_file_path,
749 path_in_repo=remote_path,
750 repo_id=self._repository_id,
751 repo_type=self._repo_type,
752 revision=self._repo_revision,
753 commit_message=f"Upload {remote_path}",
754 create_pr=False,
755 )
756
757 return len(content_bytes)
758
759 finally:
760 os.unlink(temp_file_path)
761
762 return self._translate_errors(_invoke_api, "PUT", self._repository_id, remote_path)
763
764 def _download_file(self, remote_path: str, f: str | IO, metadata: ObjectMetadata | None = None) -> int:
765 """
766 Downloads a file from the HuggingFace repository.
767
768 :param remote_path: The remote path of the file to download from the repository.
769 :param f: Local file path or file object to write to.
770 :param metadata: Optional object metadata (not used in this implementation).
771 :return: Data size in bytes.
772 """
773 if not self._hf_client:
774 raise RuntimeError("HuggingFace client not initialized")
775
776 remote_path = self._normalize_path(remote_path)
777
778 def _invoke_api():
779 if isinstance(f, str):
780 parent_dir = os.path.dirname(f)
781 if parent_dir:
782 safe_makedirs(parent_dir)
783
784 target_dir = parent_dir if parent_dir else "."
785 downloaded_path = self._hf_client.hf_hub_download(
786 repo_id=self._repository_id,
787 filename=remote_path,
788 repo_type=self._repo_type,
789 revision=self._repo_revision,
790 local_dir=target_dir,
791 )
792
793 if os.path.abspath(downloaded_path) != os.path.abspath(f):
794 os.rename(downloaded_path, f)
795
796 return os.path.getsize(f)
797
798 else:
799 with tempfile.TemporaryDirectory() as temp_dir:
800 downloaded_path = self._hf_client.hf_hub_download(
801 repo_id=self._repository_id,
802 filename=remote_path,
803 repo_type=self._repo_type,
804 revision=self._repo_revision,
805 local_dir=temp_dir,
806 )
807
808 with open(downloaded_path, "rb") as src:
809 data = src.read()
810 if isinstance(f, io.TextIOBase):
811 f.write(data.decode("utf-8"))
812 else:
813 f.write(data)
814
815 return len(data)
816
817 return self._translate_errors(_invoke_api, "GET", self._repository_id, remote_path)
818
819 def _is_dir(self, path: str) -> bool:
820 """
821 Helper method to check if a path is a directory.
822
823 :param path: The path to check.
824 :return: True if the path appears to be a directory (has files under it).
825 """
826 path = path.rstrip("/")
827 if not path:
828 # The root of the repo is always a directory
829 return True
830
831 try:
832 path_info = self._hf_client.get_paths_info(
833 repo_id=self._repository_id,
834 paths=[path],
835 repo_type=self._repo_type,
836 revision=self._repo_revision,
837 )
838
839 if not path_info:
840 return False
841
842 return isinstance(path_info[0], RepoFolder)
843
844 except RepositoryNotFoundError as e:
845 raise FileNotFoundError(
846 f"Repository not found or access denied: {self._repository_id}. "
847 f"Verify the repository exists and you have access permissions."
848 ) from e
849 except RevisionNotFoundError as e:
850 raise FileNotFoundError(
851 f"Revision '{self._repo_revision}' not found in repository {self._repository_id}. "
852 f"Verify the branch, tag, or commit exists."
853 ) from e
854 except IndexError:
855 return False
856 except Exception as e:
857 raise Exception(f"Unexpected error: {e}")
858
859 def _normalize_path(self, path: str) -> str:
860 """
861 Normalize path for HuggingFace API by removing leading slashes.
862 HuggingFace expects relative paths within repositories.
863 """
864 return path.lstrip("/")