feat(hf): add support for buckets repository type

kszucs · kszucs · commit 296979ad3c1c · 2026-02-18T11:19:29.000+01:00
diff --git a/core/services/hf/src/backend.rs b/core/services/hf/src/backend.rs
@@ -45,6 +45,7 @@ impl HfBuilder {
     /// - dataset
     /// - datasets (alias for dataset)
     /// - space
+    /// - bucket
     ///
     /// [Reference](https://huggingface.co/docs/hub/repositories)
     pub fn repo_type(mut self, repo_type: &str) -> Self {
@@ -298,6 +299,13 @@ pub(super) mod test_utils {
         (repo_id, token)
     }
 
+    #[cfg(feature = "xet")]
+    pub fn testing_bucket_credentials() -> (String, String) {
+        let repo_id = std::env::var("HF_OPENDAL_BUCKET").expect("HF_OPENDAL_BUCKET must be set");
+        let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN must be set");
+        (repo_id, token)
+    }
+
     /// Operator for a private dataset requiring HF_OPENDAL_DATASET and HF_OPENDAL_TOKEN.
     /// Uses higher max_retries to tolerate concurrent commit conflicts (412).
     pub fn testing_operator() -> Operator {
@@ -330,6 +338,24 @@ pub(super) mod test_utils {
         finish_operator(op)
     }
 
+    /// Operator for a bucket requiring HF_OPENDAL_BUCKET and HF_OPENDAL_TOKEN.
+    /// Buckets always use XET for writes.
+    #[cfg(feature = "xet")]
+    pub fn testing_bucket_operator() -> Operator {
+        let (repo_id, token) = testing_bucket_credentials();
+        let op = Operator::new(
+            HfBuilder::default()
+                .repo_type("bucket")
+                .repo_id(&repo_id)
+                .token(&token)
+                .enable_xet()
+                .max_retries(10),
+        )
+        .unwrap()
+        .finish();
+        finish_operator(op)
+    }
+
     pub fn gpt2_operator() -> Operator {
         let op = Operator::new(
             HfBuilder::default()
diff --git a/core/services/hf/src/core.rs b/core/services/hf/src/core.rs
@@ -86,6 +86,18 @@ pub(super) struct DeletedFile {
     pub path: String,
 }
 
+/// Bucket batch operation payload structures
+#[cfg(feature = "xet")]
+#[derive(Debug, serde::Serialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub(super) enum BucketOperation {
+    #[serde(rename_all = "camelCase")]
+    AddFile { path: String, xet_hash: String },
+    #[serde(rename_all = "camelCase")]
+    #[allow(dead_code)]
+    DeleteFile { path: String },
+}
+
 #[derive(serde::Serialize)]
 pub(super) struct MixedCommitPayload {
     pub summary: String,
@@ -113,7 +125,8 @@ pub(super) struct CommitResponse {
 pub(super) struct PathInfo {
     #[serde(rename = "type")]
     pub type_: String,
-    pub oid: String,
+    #[serde(default)]
+    pub oid: Option<String>,
     pub size: u64,
     #[serde(default)]
     pub lfs: Option<LfsInfo>,
@@ -141,12 +154,12 @@ impl PathInfo {
 
         if mode == EntryMode::FILE {
             meta.set_content_length(self.size);
-            let etag = if let Some(lfs) = &self.lfs {
-                &lfs.oid
-            } else {
-                &self.oid
-            };
-            meta.set_etag(etag);
+            // For buckets, oid may be None; for regular repos, prefer lfs.oid then oid
+            if let Some(lfs) = &self.lfs {
+                meta.set_etag(&lfs.oid);
+            } else if let Some(oid) = &self.oid {
+                meta.set_etag(oid);
+            }
         }
 
         Ok(meta)
@@ -511,6 +524,46 @@ impl HfCore {
         let (_, resp) = self.send_parse::<CommitResponse>(req).await?;
         Ok(resp)
     }
+
+    /// Upload files to a bucket using the batch API.
+    ///
+    /// Sends operations as JSON lines (one operation per line).
+    #[cfg(feature = "xet")]
+    pub(super) async fn bucket_batch(&self, operations: Vec<BucketOperation>) -> Result<()> {
+        let _token = self.token.as_deref().ok_or_else(|| {
+            Error::new(
+                ErrorKind::PermissionDenied,
+                "token is required for bucket operations",
+            )
+            .with_operation("bucket_batch")
+        })?;
+
+        if operations.is_empty() {
+            return Err(Error::new(
+                ErrorKind::Unexpected,
+                "no operations to perform",
+            ));
+        }
+
+        let url = self.repo.bucket_batch_url(&self.endpoint);
+
+        let mut body = String::new();
+        for op in operations {
+            let json = serde_json::to_string(&op).map_err(new_json_serialize_error)?;
+            body.push_str(&json);
+            body.push('\n');
+        }
+
+        let req = self
+            .request(http::Method::POST, &url, Operation::Write)
+            .header(header::CONTENT_TYPE, "application/x-ndjson")
+            .header(header::CONTENT_LENGTH, body.len())
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+
+        self.send(req).await?;
+        Ok(())
+    }
 }
 
 #[cfg(test)]
diff --git a/core/services/hf/src/reader.rs b/core/services/hf/src/reader.rs
@@ -27,6 +27,8 @@ use futures::StreamExt;
 use super::core::HfCore;
 #[cfg(feature = "xet")]
 use super::core::map_xet_error;
+#[cfg(feature = "xet")]
+use super::uri::RepoType;
 use opendal_core::raw::*;
 use opendal_core::*;
 #[cfg(feature = "xet")]
@@ -43,12 +45,24 @@ pub enum HfReader {
 impl HfReader {
     /// Create a reader, automatically choosing between XET and HTTP.
     ///
-    /// When XET is enabled a HEAD request probes for the `X-Xet-Hash`
-    /// header. Files stored on XET are downloaded via the CAS protocol;
-    /// all others fall back to a regular HTTP GET.
+    /// Buckets always use XET. For other repo types, when XET is enabled
+    /// a HEAD request probes for the `X-Xet-Hash` header. Files stored on
+    /// XET are downloaded via the CAS protocol; all others fall back to HTTP GET.
     pub async fn try_new(core: &HfCore, path: &str, range: BytesRange) -> Result<Self> {
         #[cfg(feature = "xet")]
         if core.xet_enabled {
+            // Buckets always use XET
+            if core.repo.repo_type == RepoType::Bucket {
+                if let Some(xet_file) = core.maybe_xet_file(path).await? {
+                    return Self::try_new_xet(core, &xet_file, range).await;
+                }
+                return Err(Error::new(
+                    ErrorKind::Unexpected,
+                    "bucket file is missing XET metadata",
+                ));
+            }
+
+            // For other repos, probe for XET
             if let Some(xet_file) = core.maybe_xet_file(path).await? {
                 return Self::try_new_xet(core, &xet_file, range).await;
             }
diff --git a/core/services/hf/src/uri.rs b/core/services/hf/src/uri.rs
@@ -22,7 +22,7 @@ use serde::Serialize;
 use super::HUGGINGFACE_SCHEME;
 use opendal_core::raw::*;
 
-/// Repository type of Huggingface. Supports `model`, `dataset`, and `space`.
+/// Repository type of Huggingface. Supports `model`, `dataset`, `space`, and `bucket`.
 /// [Reference](https://huggingface.co/docs/hub/repositories)
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[serde(rename_all = "lowercase")]
@@ -31,6 +31,7 @@ pub enum RepoType {
     Model,
     Dataset,
     Space,
+    Bucket,
 }
 
 impl RepoType {
@@ -39,6 +40,7 @@ impl RepoType {
             "model" | "models" => Ok(Self::Model),
             "dataset" | "datasets" => Ok(Self::Dataset),
             "space" | "spaces" => Ok(Self::Space),
+            "bucket" | "buckets" => Ok(Self::Bucket),
             other => Err(opendal_core::Error::new(
                 opendal_core::ErrorKind::ConfigInvalid,
                 format!("unknown repo type: {other}"),
@@ -52,6 +54,7 @@ impl RepoType {
             Self::Model => "model",
             Self::Dataset => "dataset",
             Self::Space => "space",
+            Self::Bucket => "bucket",
         }
     }
 
@@ -60,6 +63,7 @@ impl RepoType {
             Self::Model => "models",
             Self::Dataset => "datasets",
             Self::Space => "spaces",
+            Self::Bucket => "buckets",
         }
     }
 }
@@ -98,42 +102,49 @@ impl HfRepo {
 
     /// Build the paths-info API URL for this repository.
     pub fn paths_info_url(&self, endpoint: &str) -> String {
-        format!(
-            "{}/api/{}/{}/paths-info/{}",
-            endpoint,
-            self.repo_type.as_plural_str(),
-            &self.repo_id,
-            percent_encode_revision(self.revision()),
-        )
-    }
-
-    /// Build the Git LFS batch API URL for this repository.
-    ///
-    /// Pattern: `{endpoint}/{type_prefix}{repo_id}.git/info/lfs/objects/batch`
-    /// where type_prefix is "" for models, "datasets/" for datasets, "spaces/" for spaces.
-    pub fn lfs_batch_url(&self, endpoint: &str) -> String {
-        let type_prefix = match self.repo_type {
-            RepoType::Model => "",
-            RepoType::Dataset => "datasets/",
-            RepoType::Space => "spaces/",
-        };
-        format!(
-            "{}/{}{}.git/info/lfs/objects/batch",
-            endpoint, type_prefix, &self.repo_id,
-        )
+        match self.repo_type {
+            RepoType::Bucket => {
+                format!("{}/api/buckets/{}/paths-info", endpoint, &self.repo_id)
+            }
+            _ => {
+                format!(
+                    "{}/api/{}/{}/paths-info/{}",
+                    endpoint,
+                    self.repo_type.as_plural_str(),
+                    &self.repo_id,
+                    percent_encode_revision(self.revision()),
+                )
+            }
+        }
     }
 
     /// Build the XET token API URL for this repository.
     #[cfg(feature = "xet")]
     pub fn xet_token_url(&self, endpoint: &str, token_type: &str) -> String {
-        format!(
-            "{}/api/{}/{}/xet-{}-token/{}",
-            endpoint,
-            self.repo_type.as_plural_str(),
-            &self.repo_id,
-            token_type,
-            self.revision(),
-        )
+        match self.repo_type {
+            RepoType::Bucket => {
+                format!(
+                    "{}/api/buckets/{}/xet-{}-token",
+                    endpoint, &self.repo_id, token_type
+                )
+            }
+            _ => {
+                format!(
+                    "{}/api/{}/{}/xet-{}-token/{}",
+                    endpoint,
+                    self.repo_type.as_plural_str(),
+                    &self.repo_id,
+                    token_type,
+                    self.revision(),
+                )
+            }
+        }
+    }
+
+    /// Build the bucket batch API URL for this repository.
+    #[cfg(feature = "xet")]
+    pub fn bucket_batch_url(&self, endpoint: &str) -> String {
+        format!("{}/api/buckets/{}/batch", endpoint, &self.repo_id)
     }
 }
 
@@ -275,6 +286,12 @@ impl HfUri {
                     endpoint, &self.repo.repo_id, revision, path
                 )
             }
+            RepoType::Bucket => {
+                format!(
+                    "{}/buckets/{}/resolve/{}",
+                    endpoint, &self.repo.repo_id, path
+                )
+            }
         }
     }
 
@@ -535,4 +552,56 @@ mod tests {
         assert!(p.repo.revision.is_none());
         assert_eq!(p.path, "");
     }
+
+    #[test]
+    fn resolve_buckets_prefix() {
+        let p = resolve("buckets/username/my_bucket");
+        assert_eq!(p.repo.repo_type, RepoType::Bucket);
+        assert_eq!(p.repo.repo_id, "username/my_bucket");
+        assert!(p.repo.revision.is_none());
+        assert_eq!(p.path, "");
+    }
+
+    #[test]
+    fn resolve_buckets_with_path() {
+        let p = resolve("buckets/username/my_bucket/data/file.txt");
+        assert_eq!(p.repo.repo_type, RepoType::Bucket);
+        assert_eq!(p.repo.repo_id, "username/my_bucket");
+        assert!(p.repo.revision.is_none());
+        assert_eq!(p.path, "data/file.txt");
+    }
+
+    #[test]
+    fn test_bucket_resolve_url() {
+        let p = resolve("buckets/user/bucket/file.txt");
+        let url = p.resolve_url("https://huggingface.co");
+        assert_eq!(
+            url,
+            "https://huggingface.co/buckets/user/bucket/resolve/file.txt"
+        );
+    }
+
+    #[test]
+    #[cfg(feature = "xet")]
+    fn test_bucket_xet_token_urls() {
+        let p = resolve("buckets/user/bucket");
+        let read_url = p.repo.xet_token_url("https://huggingface.co", "read");
+        let write_url = p.repo.xet_token_url("https://huggingface.co", "write");
+        assert_eq!(
+            read_url,
+            "https://huggingface.co/api/buckets/user/bucket/xet-read-token"
+        );
+        assert_eq!(
+            write_url,
+            "https://huggingface.co/api/buckets/user/bucket/xet-write-token"
+        );
+    }
+
+    #[test]
+    #[cfg(feature = "xet")]
+    fn test_bucket_batch_url() {
+        let p = resolve("buckets/user/bucket");
+        let url = p.repo.bucket_batch_url("https://huggingface.co");
+        assert_eq!(url, "https://huggingface.co/api/buckets/user/bucket/batch");
+    }
 }
diff --git a/core/services/hf/src/writer.rs b/core/services/hf/src/writer.rs
