1use std::path::PathBuf;
19use std::sync::Arc;
20
21use log::debug;
22
23use super::HF_SCHEME;
24use super::config::HfConfig;
25use super::core::HfCore;
26use super::core::HfDownloadMode;
27use super::core::{HfRepo, HfRepoType};
28use super::deleter::HfDeleter;
29use super::lister::HfLister;
30use super::reader::*;
31use super::writer::HfLazyWriter;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35#[doc = include_str!("docs.md")]
37#[derive(Debug, Default)]
38pub struct HfBuilder {
39 pub(super) config: HfConfig,
40}
41
42impl HfBuilder {
43 pub fn repo_type(mut self, repo_type: &str) -> Self {
54 if !repo_type.is_empty()
55 && let Ok(rt) = HfRepoType::parse(repo_type)
56 {
57 self.config.repo_type = Some(rt);
58 }
59 self
60 }
61
62 pub fn repo_id(mut self, repo_id: &str) -> Self {
72 if !repo_id.is_empty() {
73 self.config.repo_id = Some(repo_id.to_string());
74 }
75 self
76 }
77
78 pub fn revision(mut self, revision: &str) -> Self {
86 if !revision.is_empty() {
87 self.config.revision = Some(revision.to_string());
88 }
89 self
90 }
91
92 pub fn root(mut self, root: &str) -> Self {
96 self.config.root = if root.is_empty() {
97 None
98 } else {
99 Some(root.to_string())
100 };
101
102 self
103 }
104
105 pub fn token(mut self, token: &str) -> Self {
109 if !token.is_empty() {
110 self.config.token = Some(token.to_string());
111 }
112 self
113 }
114
115 pub fn download_mode(mut self, mode: &str) -> Self {
128 if !mode.is_empty()
129 && let Ok(m) = HfDownloadMode::parse(mode)
130 {
131 self.config.download_mode = Some(m);
132 }
133 self
134 }
135
136 pub fn endpoint(mut self, endpoint: &str) -> Self {
143 if !endpoint.is_empty() {
144 self.config.endpoint = Some(endpoint.to_string());
145 }
146 self
147 }
148
149 fn hf_endpoint(&self) -> String {
150 self.config
151 .endpoint
152 .clone()
153 .or_else(|| std::env::var("HF_ENDPOINT").ok())
154 .unwrap_or_else(|| "https://huggingface.co".to_string())
155 }
156
157 fn hf_download_mode(&self) -> HfDownloadMode {
160 if let Some(mode) = self.config.download_mode {
161 return mode;
162 }
163 if let Ok(val) = std::env::var("HF_HUB_DISABLE_XET")
164 && !val.is_empty()
165 {
166 return HfDownloadMode::Http;
167 }
168 HfDownloadMode::default()
169 }
170
171 fn hf_home() -> Option<PathBuf> {
172 if let Ok(h) = std::env::var("HF_HOME") {
173 return Some(PathBuf::from(h));
174 }
175 if let Ok(xdg) = std::env::var("XDG_CACHE_HOME") {
176 return Some(PathBuf::from(xdg).join("huggingface"));
177 }
178 let home = std::env::var("HOME").ok()?;
179 Some(PathBuf::from(home).join(".cache/huggingface"))
180 }
181
182 fn hf_token(&self) -> Option<String> {
185 if let Some(t) = self.config.token.clone() {
186 return Some(t);
187 }
188 if let Ok(val) = std::env::var("HF_HUB_DISABLE_IMPLICIT_TOKEN")
189 && !val.is_empty()
190 {
191 return None;
192 }
193 if let Ok(t) = std::env::var("HF_TOKEN")
194 && !t.is_empty()
195 {
196 return Some(t);
197 }
198 let token_path = if let Ok(p) = std::env::var("HF_TOKEN_PATH") {
199 Some(PathBuf::from(p))
200 } else {
201 Self::hf_home().map(|h| h.join("token"))
202 };
203 token_path
204 .and_then(|p| std::fs::read_to_string(p).ok())
205 .map(|s| s.trim().to_string())
206 .filter(|s| !s.is_empty())
207 }
208}
209
210impl Builder for HfBuilder {
211 type Config = HfConfig;
212
213 fn build(self) -> Result<impl Service> {
214 debug!("backend build started: {:?}", self);
215
216 let token = self.hf_token();
217 let endpoint = self.hf_endpoint();
218 let download_mode = self.hf_download_mode();
219
220 let repo_type = self.config.repo_type.ok_or_else(|| {
221 Error::new(ErrorKind::ConfigInvalid, "repo_type is required")
222 .with_operation("Builder::build")
223 .with_context("service", HF_SCHEME)
224 })?;
225 debug!("backend use repo_type: {:?}", repo_type);
226
227 let repo_id = self.config.repo_id.ok_or_else(|| {
228 Error::new(ErrorKind::ConfigInvalid, "repo_id is required")
229 .with_operation("Builder::build")
230 .with_context("service", HF_SCHEME)
231 })?;
232 debug!("backend use repo_id: {}", repo_id);
233
234 let revision = match &self.config.revision {
235 Some(revision) => revision.clone(),
236 None => "main".to_string(),
237 };
238 debug!("backend use revision: {}", revision);
239
240 let root = normalize_root(&self.config.root.unwrap_or_default());
241 debug!("backend use root: {}", root);
242 debug!("backend use token: {}", token.is_some());
243 debug!("backend use endpoint: {}", endpoint);
244 debug!("backend use download_mode: {:?}", download_mode);
245
246 let info = ServiceInfo::new(HF_SCHEME, "", "");
247 let capability = Capability {
248 stat: true,
249 read: true,
250 write: token.is_some(),
251 delete: token.is_some(),
252 delete_max_size: Some(100),
253 list: true,
254 list_with_recursive: true,
255 shared: true,
256 ..Default::default()
257 };
258
259 let repo = HfRepo::new(repo_type, repo_id, Some(revision.clone()));
260 debug!("backend repo uri: {:?}", repo.uri(&root, ""));
261
262 Ok(HfBackend {
263 core: Arc::new(HfCore::build(
264 info,
265 capability,
266 repo,
267 root,
268 token,
269 endpoint,
270 download_mode,
271 )?),
272 })
273 }
274}
275
276#[derive(Debug, Clone)]
278pub struct HfBackend {
279 pub(crate) core: Arc<HfCore>,
280}
281
282impl Service for HfBackend {
283 type Reader = oio::StreamReader<HfReader>;
284 type Writer = HfLazyWriter;
285 type Lister = oio::PageLister<HfLister>;
286 type Deleter = oio::BatchDeleter<HfDeleter>;
287 type Copier = ();
288
289 fn info(&self) -> ServiceInfo {
290 self.core.info.clone()
291 }
292
293 fn capability(&self) -> Capability {
294 self.core.capability
295 }
296
297 async fn create_dir(
298 &self,
299 _ctx: &OperationContext,
300 _path: &str,
301 _args: OpCreateDir,
302 ) -> Result<RpCreateDir> {
303 Err(Error::new(
304 ErrorKind::Unsupported,
305 "operation is not supported",
306 ))
307 }
308
309 async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
310 if path == "/" {
312 return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
313 }
314
315 if self.core.repo.is_bucket() && path.ends_with('/') {
317 return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
318 }
319
320 let info = self.core.path_info(ctx, path).await?;
321 Ok(RpStat::new(info.metadata()?))
322 }
323 fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
324 let output: oio::StreamReader<HfReader> = {
325 Ok(oio::StreamReader::new(HfReader::new(
326 self.clone(),
327 ctx.clone(),
328 path,
329 args,
330 )))
331 }?;
332
333 Ok(output)
334 }
335
336 fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
337 let output: oio::PageLister<HfLister> = {
338 let lister = HfLister::new(
339 self.core.clone(),
340 ctx.clone(),
341 path.to_string(),
342 args.recursive(),
343 );
344 Ok(oio::PageLister::new(lister))
345 }?;
346
347 Ok(output)
348 }
349
350 fn write(&self, ctx: &OperationContext, path: &str, _args: OpWrite) -> Result<Self::Writer> {
351 Ok(HfLazyWriter::new(
352 self.core.clone(),
353 ctx.clone(),
354 path.to_string(),
355 ))
356 }
357
358 fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
359 let output: oio::BatchDeleter<HfDeleter> = {
360 let deleter = HfDeleter::new(self.core.clone(), ctx.clone());
361 let max_batch_size = self.core.capability.delete_max_size;
362 Ok(oio::BatchDeleter::new(deleter, max_batch_size))
363 }?;
364
365 Ok(output)
366 }
367
368 fn copy(
369 &self,
370 _ctx: &OperationContext,
371 _from: &str,
372 _to: &str,
373 _args: OpCopy,
374 _opts: OpCopier,
375 ) -> Result<Self::Copier> {
376 Err(Error::new(
377 ErrorKind::Unsupported,
378 "operation is not supported",
379 ))
380 }
381
382 async fn rename(
383 &self,
384 _ctx: &OperationContext,
385 _from: &str,
386 _to: &str,
387 _args: OpRename,
388 ) -> Result<RpRename> {
389 Err(Error::new(
390 ErrorKind::Unsupported,
391 "operation is not supported",
392 ))
393 }
394
395 async fn presign(
396 &self,
397 _ctx: &OperationContext,
398 _path: &str,
399 _args: OpPresign,
400 ) -> Result<RpPresign> {
401 Err(Error::new(
402 ErrorKind::Unsupported,
403 "operation is not supported",
404 ))
405 }
406}
407
408#[cfg(test)]
409pub(super) mod test_utils {
410 use std::sync::Arc;
411
412 use super::super::core::{HfCore, HfDownloadMode};
413 use super::super::core::{HfRepo, HfRepoType};
414 use super::HfBuilder;
415 use opendal_core::Capability;
416 use opendal_core::HttpTransporter;
417 use opendal_core::OperationContext;
418 use opendal_core::Operator;
419 use opendal_core::raw::ServiceInfo;
420
421 fn finish_operator(op: Operator) -> Operator {
422 let transport =
423 HttpTransporter::new(opendal_http_transport_reqwest::ReqwestTransport::default());
424 op.with_context(OperationContext::new().with_http_transport(transport))
425 }
426
427 pub fn mbpp_operator() -> Operator {
428 let op = Operator::new(
429 HfBuilder::default()
430 .repo_type("dataset")
431 .repo_id("google-research-datasets/mbpp"),
432 )
433 .unwrap();
434 finish_operator(op)
435 }
436
437 pub fn testing_dataset_core() -> Arc<HfCore> {
438 let repo_id = std::env::var("HF_OPENDAL_DATASET").expect("HF_OPENDAL_DATASET must be set");
439 let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN must be set");
440
441 let info = ServiceInfo::new("hf", "", "");
442 let capability = Capability {
443 read: true,
444 write: true,
445 delete: true,
446 ..Default::default()
447 };
448
449 let repo = HfRepo::new(HfRepoType::Dataset, repo_id, Some("main".to_string()));
450
451 Arc::new(
452 HfCore::build(
453 info,
454 capability,
455 repo,
456 "/".to_string(),
457 Some(token),
458 "https://huggingface.co".to_string(),
459 HfDownloadMode::Xet,
460 )
461 .expect("failed to build HfCore"),
462 )
463 }
464
465 pub fn testing_bucket_operator() -> Operator {
466 let repo_id = std::env::var("HF_OPENDAL_BUCKET").expect("HF_OPENDAL_BUCKET must be set");
467 let token = std::env::var("HF_OPENDAL_TOKEN").expect("HF_OPENDAL_TOKEN must be set");
468 let op = Operator::new(
469 HfBuilder::default()
470 .repo_type("bucket")
471 .repo_id(&repo_id)
472 .token(&token),
473 )
474 .unwrap();
475 finish_operator(op)
476 }
477}
478
479#[cfg(test)]
480mod tests {
481 use super::*;
482 use std::sync::Mutex;
483
484 static ENV_LOCK: Mutex<()> = Mutex::new(());
486
487 fn builder_with_token(token: &str) -> HfBuilder {
488 HfBuilder::default().token(token)
489 }
490
491 fn builder_no_token() -> HfBuilder {
492 HfBuilder::default()
493 }
494
495 #[test]
496 fn hf_token_config_takes_priority_over_env() {
497 let _guard = ENV_LOCK.lock().unwrap();
498 unsafe { std::env::set_var("HF_TOKEN", "env-token") };
499 let result = builder_with_token("config-token").hf_token();
500 unsafe { std::env::remove_var("HF_TOKEN") };
501 assert_eq!(result.as_deref(), Some("config-token"));
502 }
503
504 #[test]
505 fn hf_token_reads_hf_token_env_var() {
506 let _guard = ENV_LOCK.lock().unwrap();
507 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
508 unsafe { std::env::remove_var("HF_TOKEN_PATH") };
509 unsafe { std::env::set_var("HF_TOKEN", "my-env-token") };
510 let result = builder_no_token().hf_token();
511 unsafe { std::env::remove_var("HF_TOKEN") };
512 assert_eq!(result.as_deref(), Some("my-env-token"));
513 }
514
515 #[test]
516 fn hf_token_disable_flag_suppresses_discovery() {
517 let _guard = ENV_LOCK.lock().unwrap();
518 unsafe { std::env::set_var("HF_HUB_DISABLE_IMPLICIT_TOKEN", "1") };
519 unsafe { std::env::set_var("HF_TOKEN", "my-env-token") };
520 let result = builder_no_token().hf_token();
521 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
522 unsafe { std::env::remove_var("HF_TOKEN") };
523 assert_eq!(result, None);
524 }
525
526 #[test]
527 fn hf_token_reads_from_file_via_hf_token_path() {
528 let _guard = ENV_LOCK.lock().unwrap();
529 let token_file = std::env::temp_dir().join("opendal-hf-token-test");
530 std::fs::write(&token_file, "file-token\n").unwrap();
531 unsafe { std::env::remove_var("HF_HUB_DISABLE_IMPLICIT_TOKEN") };
532 unsafe { std::env::remove_var("HF_TOKEN") };
533 unsafe { std::env::set_var("HF_TOKEN_PATH", &token_file) };
534 let result = builder_no_token().hf_token();
535 unsafe { std::env::remove_var("HF_TOKEN_PATH") };
536 std::fs::remove_file(&token_file).ok();
537 assert_eq!(result.as_deref(), Some("file-token"));
538 }
539
540 #[test]
541 fn hf_endpoint_returns_default() {
542 let _guard = ENV_LOCK.lock().unwrap();
543 unsafe { std::env::remove_var("HF_ENDPOINT") };
544 let result = HfBuilder::default().hf_endpoint();
545 assert_eq!(result, "https://huggingface.co");
546 }
547
548 #[test]
549 fn hf_endpoint_config_takes_priority_over_env() {
550 let _guard = ENV_LOCK.lock().unwrap();
551 unsafe { std::env::set_var("HF_ENDPOINT", "https://env.example.com") };
552 let result = HfBuilder::default()
553 .endpoint("https://config.example.com")
554 .hf_endpoint();
555 unsafe { std::env::remove_var("HF_ENDPOINT") };
556 assert_eq!(result, "https://config.example.com");
557 }
558
559 #[test]
560 fn hf_endpoint_reads_hf_endpoint_env_var() {
561 let _guard = ENV_LOCK.lock().unwrap();
562 unsafe { std::env::set_var("HF_ENDPOINT", "https://env.example.com") };
563 let result = HfBuilder::default().hf_endpoint();
564 unsafe { std::env::remove_var("HF_ENDPOINT") };
565 assert_eq!(result, "https://env.example.com");
566 }
567
568 #[test]
569 fn hf_download_mode_defaults_to_xet() {
570 let _guard = ENV_LOCK.lock().unwrap();
571 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
572 assert_eq!(HfBuilder::default().hf_download_mode(), HfDownloadMode::Xet);
573 }
574
575 #[test]
576 fn hf_download_mode_disable_xet_env_forces_http() {
577 let _guard = ENV_LOCK.lock().unwrap();
578 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "1") };
579 let mode = HfBuilder::default().hf_download_mode();
580 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
581 assert_eq!(mode, HfDownloadMode::Http);
582 }
583
584 #[test]
585 fn hf_download_mode_config_takes_priority_over_env() {
586 let _guard = ENV_LOCK.lock().unwrap();
587 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "1") };
588 let mode = HfBuilder::default().download_mode("xet").hf_download_mode();
589 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
590 assert_eq!(mode, HfDownloadMode::Xet);
591 }
592
593 #[test]
594 fn hf_download_mode_empty_env_keeps_xet() {
595 let _guard = ENV_LOCK.lock().unwrap();
596 unsafe { std::env::set_var("HF_HUB_DISABLE_XET", "") };
597 let mode = HfBuilder::default().hf_download_mode();
598 unsafe { std::env::remove_var("HF_HUB_DISABLE_XET") };
599 assert_eq!(mode, HfDownloadMode::Xet);
600 }
601
602 #[test]
603 fn build_accepts_datasets_alias() {
604 HfBuilder::default()
605 .repo_id("org/repo")
606 .repo_type("datasets")
607 .build()
608 .expect("builder should accept datasets alias");
609 }
610
611 #[test]
612 fn build_accepts_space_repo_type() {
613 HfBuilder::default()
614 .repo_id("org/space")
615 .repo_type("space")
616 .build()
617 .expect("builder should accept space repo type");
618 }
619
620 #[test]
621 fn test_both_schemes_are_supported() {
622 use opendal_core::OperatorRegistry;
623
624 let registry = OperatorRegistry::get();
625 super::super::register_hf_service(registry);
626
627 let op = registry
629 .load("hf://user/repo")
630 .expect("short scheme should be registered and work");
631 assert_eq!(op.info().scheme(), "hf");
632
633 let op = registry
635 .load("huggingface://user/repo")
636 .expect("long scheme should be registered and work");
637 assert_eq!(op.info().scheme(), "hf");
638 }
639}