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 # expand=True populates last_commit; fall back to the epoch sentinel when it is absent.
520 last_modified = AWARE_DATETIME_MIN
521 last_commit = getattr(item, "last_commit", None)
522 if last_commit is not None and getattr(last_commit, "date", None):
523 last_modified = last_commit.date
524
525 if isinstance(item, RepoFile):
526 etag = item.blob_id
527 return ObjectMetadata(
528 key=item.path,
529 type="file",
530 content_length=item.size,
531 last_modified=last_modified,
532 etag=etag,
533 content_type=None,
534 storage_class=None,
535 metadata=None,
536 )
537 else:
538 etag = item.tree_id
539 return ObjectMetadata(
540 key=item.path,
541 type="directory",
542 content_length=0,
543 last_modified=last_modified,
544 etag=etag,
545 content_type=None,
546 storage_class=None,
547 metadata=None,
548 )
549
550 def _make_symlink(self, path: str, target: str) -> None:
551 """
552 Not supported. HuggingFace repositories are read-only through this provider.
553
554 :raises NotImplementedError: Always.
555 """
556 raise NotImplementedError("HuggingFace provider does not support symlink creation.")
557
558 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
559 """
560 Retrieves metadata for an object in the HuggingFace repository.
561
562 :param path: The path of the object to get metadata for.
563 :param strict: Whether to raise an error if the object doesn't exist.
564 :return: Metadata about the object.
565 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
566 :raises FileNotFoundError: If the file doesn't exist and strict=True.
567 """
568 if not self._hf_client:
569 raise RuntimeError("HuggingFace client not initialized")
570
571 path = self._normalize_path(path)
572
573 def _invoke_api():
574 items = self._hf_client.get_paths_info(
575 repo_id=self._repository_id,
576 paths=[path],
577 repo_type=self._repo_type,
578 revision=self._repo_revision,
579 expand=True,
580 )
581
582 if not items:
583 raise FileNotFoundError(f"File not found in HuggingFace repository: {path}")
584
585 item = items[0]
586 return self._item_to_metadata(item)
587
588 try:
589 return self._translate_errors(_invoke_api, "HEAD", self._repository_id, path)
590 except FileNotFoundError:
591 if strict:
592 dir_path = path.rstrip("/") + "/"
593 if self._is_dir(dir_path):
594 return ObjectMetadata(
595 key=dir_path,
596 type="directory",
597 content_length=0,
598 last_modified=AWARE_DATETIME_MIN,
599 etag=None,
600 content_type=None,
601 storage_class=None,
602 metadata=None,
603 )
604 raise
605
606 def _list_objects(
607 self,
608 path: str,
609 start_after: str | None = None,
610 end_at: str | None = None,
611 include_directories: bool = False,
612 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
613 ) -> Iterator[ObjectMetadata]:
614 """
615 Lists objects in the HuggingFace repository under the specified path.
616
617 :param path: The path to list objects under.
618 :param start_after: The key to start listing after (exclusive, used as cursor).
619 :param end_at: The key to end listing at (inclusive, used as cursor).
620 :param include_directories: Whether to include directories in the listing.
621 :return: An iterator over object metadata for objects under the specified path.
622 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
623
624 .. note::
625 HuggingFace Hub API does not natively support pagination parameters.
626 This implementation fetches all items and uses cursor-based filtering,
627 which may impact performance for large repositories. The ordering is
628 directory-first, then files, with lexicographical ordering within each group.
629 """
630 if not self._hf_client:
631 raise RuntimeError("HuggingFace client not initialized")
632
633 path = self._normalize_path(path)
634
635 try:
636 metadata = self._get_object_metadata(path.rstrip("/"), strict=False)
637 if metadata and metadata.type == "file":
638 yield metadata
639 return
640 except FileNotFoundError:
641 pass
642
643 def _invoke_api():
644 dir_path = path.rstrip("/")
645
646 try:
647 repo_items = self._hf_client.list_repo_tree(
648 repo_id=self._repository_id,
649 path_in_repo=dir_path + "/" if dir_path else None,
650 repo_type=self._repo_type,
651 revision=self._repo_revision,
652 expand=True,
653 recursive=not include_directories,
654 )
655 return list(repo_items)
656 except EntryNotFoundError:
657 # Directory doesn't exist - return empty (matches POSIX behavior). A missing repository or
658 # revision is translated to FileNotFoundError by _translate_errors and propagates.
659 return []
660
661 items = self._translate_errors(_invoke_api, "LIST", self._repository_id, path)
662
663 # Use cursor-based pagination because HuggingFace returns items with
664 # directory-first ordering (not pure lexicographical).
665 seen_start = start_after is None
666 seen_end = False
667
668 for item in items:
669 if seen_end:
670 break
671
672 metadata = self._item_to_metadata(item)
673 key = metadata.key
674
675 if not seen_start:
676 if key == start_after:
677 seen_start = True
678 continue
679
680 should_yield = False
681 if include_directories and isinstance(item, RepoFolder) or isinstance(item, RepoFile):
682 should_yield = True
683
684 if should_yield:
685 yield metadata
686
687 if end_at is not None and key == end_at:
688 seen_end = True
689
690 def _upload_file(self, remote_path: str, f: str | IO, attributes: dict[str, str] | None = None) -> int:
691 """
692 Uploads a file to the HuggingFace repository.
693
694 :param remote_path: The remote path where the file will be stored in the repository.
695 :param f: File path or file object to upload.
696 :param attributes: Optional attributes for the file (not supported by HuggingFace).
697 :return: Data size in bytes.
698 :raises RuntimeError: If HuggingFace client is not initialized or API errors occur.
699 :raises ValueError: If client attempts to create a directory.
700 :raises ValueError: If custom attributes are provided (not supported).
701 """
702 if not self._hf_client:
703 raise RuntimeError("HuggingFace client not initialized")
704
705 if attributes is not None:
706 raise ValueError(
707 "HuggingFace provider does not support custom file attributes. "
708 "Use commit messages or repository metadata instead."
709 )
710
711 if remote_path.endswith("/"):
712 raise ValueError(
713 "HuggingFace Storage Provider does not support explicit directory creation. "
714 "Directories are created implicitly when files are uploaded to paths within them."
715 )
716
717 remote_path = self._normalize_path(remote_path)
718
719 def _invoke_api():
720 if isinstance(f, str):
721 file_size = os.path.getsize(f)
722
723 self._hf_client.upload_file(
724 path_or_fileobj=f,
725 path_in_repo=remote_path,
726 repo_id=self._repository_id,
727 repo_type=self._repo_type,
728 revision=self._repo_revision,
729 commit_message=f"Upload {remote_path}",
730 commit_description=None,
731 create_pr=False,
732 )
733
734 return file_size
735
736 else:
737 content = f.read()
738
739 if isinstance(content, str):
740 content_bytes = content.encode("utf-8")
741 else:
742 content_bytes = content
743
744 # Create temporary file since HfAPI.upload_file requires BinaryIO, not generic IO
745 with tempfile.NamedTemporaryFile(delete=False) as temp_file:
746 temp_file.write(content_bytes)
747 temp_file_path = temp_file.name
748
749 try:
750 self._hf_client.upload_file(
751 path_or_fileobj=temp_file_path,
752 path_in_repo=remote_path,
753 repo_id=self._repository_id,
754 repo_type=self._repo_type,
755 revision=self._repo_revision,
756 commit_message=f"Upload {remote_path}",
757 create_pr=False,
758 )
759
760 return len(content_bytes)
761
762 finally:
763 os.unlink(temp_file_path)
764
765 return self._translate_errors(_invoke_api, "PUT", self._repository_id, remote_path)
766
767 def _download_file(self, remote_path: str, f: str | IO, metadata: ObjectMetadata | None = None) -> int:
768 """
769 Downloads a file from the HuggingFace repository.
770
771 :param remote_path: The remote path of the file to download from the repository.
772 :param f: Local file path or file object to write to.
773 :param metadata: Optional object metadata (not used in this implementation).
774 :return: Data size in bytes.
775 """
776 if not self._hf_client:
777 raise RuntimeError("HuggingFace client not initialized")
778
779 remote_path = self._normalize_path(remote_path)
780
781 def _invoke_api():
782 if isinstance(f, str):
783 parent_dir = os.path.dirname(f)
784 if parent_dir:
785 safe_makedirs(parent_dir)
786
787 target_dir = parent_dir if parent_dir else "."
788 downloaded_path = self._hf_client.hf_hub_download(
789 repo_id=self._repository_id,
790 filename=remote_path,
791 repo_type=self._repo_type,
792 revision=self._repo_revision,
793 local_dir=target_dir,
794 )
795
796 if os.path.abspath(downloaded_path) != os.path.abspath(f):
797 os.rename(downloaded_path, f)
798
799 return os.path.getsize(f)
800
801 else:
802 with tempfile.TemporaryDirectory() as temp_dir:
803 downloaded_path = self._hf_client.hf_hub_download(
804 repo_id=self._repository_id,
805 filename=remote_path,
806 repo_type=self._repo_type,
807 revision=self._repo_revision,
808 local_dir=temp_dir,
809 )
810
811 with open(downloaded_path, "rb") as src:
812 data = src.read()
813 if isinstance(f, io.TextIOBase):
814 f.write(data.decode("utf-8"))
815 else:
816 f.write(data)
817
818 return len(data)
819
820 return self._translate_errors(_invoke_api, "GET", self._repository_id, remote_path)
821
822 def _is_dir(self, path: str) -> bool:
823 """
824 Helper method to check if a path is a directory.
825
826 :param path: The path to check.
827 :return: True if the path appears to be a directory (has files under it).
828 """
829 path = path.rstrip("/")
830 if not path:
831 # The root of the repo is always a directory
832 return True
833
834 try:
835 path_info = self._hf_client.get_paths_info(
836 repo_id=self._repository_id,
837 paths=[path],
838 repo_type=self._repo_type,
839 revision=self._repo_revision,
840 )
841
842 if not path_info:
843 return False
844
845 return isinstance(path_info[0], RepoFolder)
846
847 except RepositoryNotFoundError as e:
848 raise FileNotFoundError(
849 f"Repository not found or access denied: {self._repository_id}. "
850 f"Verify the repository exists and you have access permissions."
851 ) from e
852 except RevisionNotFoundError as e:
853 raise FileNotFoundError(
854 f"Revision '{self._repo_revision}' not found in repository {self._repository_id}. "
855 f"Verify the branch, tag, or commit exists."
856 ) from e
857 except IndexError:
858 return False
859 except Exception as e:
860 raise Exception(f"Unexpected error: {e}")
861
862 def _normalize_path(self, path: str) -> str:
863 """
864 Normalize path for HuggingFace API by removing leading slashes.
865 HuggingFace expects relative paths within repositories.
866 """
867 return path.lstrip("/")